diff --git a/.github/trigger_files/beam_PostCommit_Python_Xlang_Gcp_Direct.json b/.github/trigger_files/beam_PostCommit_Python_Xlang_Gcp_Direct.json index bb5da04014ec..83346d34aee0 100644 --- a/.github/trigger_files/beam_PostCommit_Python_Xlang_Gcp_Direct.json +++ b/.github/trigger_files/beam_PostCommit_Python_Xlang_Gcp_Direct.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run", - "modification": 15 + "modification": 16 } diff --git a/.github/workflows/build_wheels.yml b/.github/workflows/build_wheels.yml index 7bbbb1a2e3db..1bd4d297a294 100644 --- a/.github/workflows/build_wheels.yml +++ b/.github/workflows/build_wheels.yml @@ -244,13 +244,13 @@ jobs: - name: Install Python uses: actions/setup-python@v5 with: - python-version: '3.10' + python-version: '3.11' - uses: docker/setup-qemu-action@v3 if: ${{matrix.os_python.arch == 'aarch64'}} name: Set up QEMU - name: Install cibuildwheel # note: sync cibuildwheel version with gradle task sdks:python:bdistPy* steps - run: pip install cibuildwheel==2.23.3 setuptools + run: pip install cibuildwheel==3.3.1 setuptools - name: Build wheel # Only build wheel if it is one of the target versions for this platform, otherwise no-op if: ${{ contains(matrix.os_python.python, matrix.py_version) }} diff --git a/.github/workflows/run_rc_validation_java_quickstart.yml b/.github/workflows/run_rc_validation_java_quickstart.yml index f39e8ac93923..98b597bf4c0e 100644 --- a/.github/workflows/run_rc_validation_java_quickstart.yml +++ b/.github/workflows/run_rc_validation_java_quickstart.yml @@ -88,7 +88,7 @@ jobs: - name: Run QuickStart Java Flink Runner uses: ./.github/actions/gradle-command-self-hosted-action with: - gradle-command: :runners:flink:1.20:runQuickstartJavaFlinkLocal + gradle-command: :runners:flink:2.0:runQuickstartJavaFlinkLocal arguments: | -Prepourl=${{ env.APACHE_REPO_URL }} \ -Pver=${{ env.RELEASE_VERSION }} diff --git a/release/build.gradle.kts b/release/build.gradle.kts index b3438ee79cdb..5be707428605 100644 --- a/release/build.gradle.kts +++ b/release/build.gradle.kts @@ -39,7 +39,7 @@ task("runJavaExamplesValidationTask") { dependsOn(":runners:direct-java:runQuickstartJavaDirect") dependsOn(":runners:google-cloud-dataflow-java:runQuickstartJavaDataflow") dependsOn(":runners:spark:3:runQuickstartJavaSpark") - dependsOn(":runners:flink:1.20:runQuickstartJavaFlinkLocal") + dependsOn(":runners:flink:2.0:runQuickstartJavaFlinkLocal") dependsOn(":runners:direct-java:runMobileGamingJavaDirect") if (project.hasProperty("ver") || !project.version.toString().endsWith("SNAPSHOT")) { // only run one variant of MobileGaming on Dataflow for nightly diff --git a/sdks/python/apache_beam/ml/rag/enrichment/bigquery_vector_search.py b/sdks/python/apache_beam/ml/rag/enrichment/bigquery_vector_search.py index 614e5f9c0800..e9269af27bd4 100644 --- a/sdks/python/apache_beam/ml/rag/enrichment/bigquery_vector_search.py +++ b/sdks/python/apache_beam/ml/rag/enrichment/bigquery_vector_search.py @@ -241,7 +241,7 @@ def format_query(self, items: List[EmbeddableItem]) -> str: ARRAY_AGG( STRUCT({"distance, " if self.include_distance else ""}\ {base_columns_str}) - ) as embeddable_items + ) as chunks FROM VECTOR_SEARCH( (SELECT {columns_str}, {self.embedding_column} FROM `{self.table_name}` diff --git a/sdks/python/apache_beam/ml/rag/ingestion/bigquery.py b/sdks/python/apache_beam/ml/rag/ingestion/bigquery.py index af170992b09c..2a7111c0d35f 100644 --- a/sdks/python/apache_beam/ml/rag/ingestion/bigquery.py +++ b/sdks/python/apache_beam/ml/rag/ingestion/bigquery.py @@ -14,8 +14,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +import warnings from collections.abc import Callable -from dataclasses import dataclass from typing import Any from typing import Dict from typing import Optional @@ -28,41 +28,56 @@ from apache_beam.typehints.row_type import RowTypeConstraint EmbeddableToDictFn = Callable[[EmbeddableItem], Dict[str, any]] +# Backward compatibility alias. +ChunkToDictFn = EmbeddableToDictFn -@dataclass class SchemaConfig: - """Configuration for custom BigQuery schema and row conversion. - - Allows overriding the default schema and row conversion logic for BigQuery - vector storage. This enables custom table schemas beyond the default - id/embedding/content/metadata structure. - - Attributes: - schema: BigQuery TableSchema dict defining the table structure. - Example: - >>> { - ... 'fields': [ - ... {'name': 'id', 'type': 'STRING'}, - ... {'name': 'embedding', 'type': 'FLOAT64', 'mode': 'REPEATED'}, - ... {'name': 'custom_field', 'type': 'STRING'} - ... ] - ... } - embeddable_to_dict_fn: Function that converts an - EmbeddableItem to a dict matching the schema. - Takes an EmbeddableItem and returns - Dict[str, Any] with keys matching - schema fields. - Example: - >>> def embeddable_to_dict(item: EmbeddableItem) -> Dict[str, Any]: - ... return { - ... 'id': item.id, - ... 'embedding': item.embedding.dense_embedding, - ... 'custom_field': item.metadata.get('custom_field') - ... } - """ - schema: Dict - embeddable_to_dict_fn: EmbeddableToDictFn + def __init__( + self, + schema: Dict, + embeddable_to_dict_fn: Optional[EmbeddableToDictFn] = None, + **kwargs): + """Configuration for custom BigQuery schema and row conversion. + + Allows overriding the default schema and row conversion logic for BigQuery + vector storage. This enables custom table schemas beyond the default + id/embedding/content/metadata structure. + + Args: + schema: BigQuery TableSchema dict defining the table structure. + embeddable_to_dict_fn: Function that converts an EmbeddableItem to a + dict matching the schema. Takes an EmbeddableItem and returns + Dict[str, Any] with keys matching schema fields. + + Example with custom schema: + >>> schema_config = SchemaConfig( + ... schema={ + ... 'fields': [ + ... {'name': 'id', 'type': 'STRING'}, + ... {'name': 'embedding', 'type': 'FLOAT64', 'mode': 'REPEATED'}, + ... {'name': 'source_url', 'type': 'STRING'} + ... ] + ... }, + ... embeddable_to_dict_fn=lambda item: { + ... 'id': item.id, + ... 'embedding': item.embedding.dense_embedding, + ... 'source_url': item.metadata.get('url') + ... } + ... ) + """ + self.schema = schema + if 'chunk_to_dict_fn' in kwargs: + warnings.warn( + "chunk_to_dict_fn is deprecated, use embeddable_to_dict_fn", + DeprecationWarning, + stacklevel=2) + embeddable_to_dict_fn = kwargs.pop('chunk_to_dict_fn') + if kwargs: + raise TypeError(f"Unexpected keyword arguments: {', '.join(kwargs)}") + if embeddable_to_dict_fn is None: + raise TypeError("SchemaConfig requires embeddable_to_dict_fn") + self.embeddable_to_dict_fn = embeddable_to_dict_fn class BigQueryVectorWriterConfig(VectorDatabaseWriteConfig): diff --git a/sdks/python/apache_beam/typehints/typehints.py b/sdks/python/apache_beam/typehints/typehints.py index d0dfaec23afc..f429935c3a0e 100644 --- a/sdks/python/apache_beam/typehints/typehints.py +++ b/sdks/python/apache_beam/typehints/typehints.py @@ -1462,9 +1462,10 @@ def normalize(x, none_as_type=False): # Convert bare builtin types to correct type hints directly elif x in _KNOWN_PRIMITIVE_TYPES: return _KNOWN_PRIMITIVE_TYPES[x] - elif getattr(x, '__module__', - None) in ('typing', 'collections', 'collections.abc') or getattr( - x, '__origin__', None) in _KNOWN_PRIMITIVE_TYPES: + elif isinstance(x, types.UnionType) or getattr( + x, '__module__', + None) in ('typing', 'collections', 'collections.abc') or getattr( + x, '__origin__', None) in _KNOWN_PRIMITIVE_TYPES: beam_type = native_type_compatibility.convert_to_beam_type(x) if beam_type != x: # We were able to do the conversion. diff --git a/sdks/python/apache_beam/typehints/typehints_test.py b/sdks/python/apache_beam/typehints/typehints_test.py index cec830380087..1377bea6d56d 100644 --- a/sdks/python/apache_beam/typehints/typehints_test.py +++ b/sdks/python/apache_beam/typehints/typehints_test.py @@ -1596,6 +1596,22 @@ def test_hint_helper(self): self.assertFalse(is_consistent_with(Union[str, int], str)) self.assertFalse(is_consistent_with(str, NonBuiltInGeneric[str])) + def test_hint_helper_pipe_union(self): + pipe_union = int | None # pylint: disable=unsupported-binary-operation + typing_union = Union[int, None] + self.assertTrue(is_consistent_with(int, pipe_union)) + self.assertTrue(is_consistent_with(type(None), pipe_union)) + self.assertFalse(is_consistent_with(str, pipe_union)) + self.assertTrue( + is_consistent_with(int, pipe_union) == is_consistent_with( + int, typing_union)) + self.assertTrue( + is_consistent_with(str, pipe_union) == is_consistent_with( + str, typing_union)) + pipe_union_2 = int | float # pylint: disable=unsupported-binary-operation + self.assertTrue(is_consistent_with(int, pipe_union_2)) + self.assertTrue(is_consistent_with(float, pipe_union_2)) + def test_positional_arg_hints(self): self.assertEqual(typehints.Any, _positional_arg_hints('x', {})) self.assertEqual(int, _positional_arg_hints('x', {'x': int})) @@ -1934,6 +1950,14 @@ def test_pipe_operator_as_union(self): native_type_compatibility.convert_to_beam_type(type_a), native_type_compatibility.convert_to_beam_type(type_b)) + def test_normalize_pipe_union(self): + pipe_union = int | None # pylint: disable=unsupported-binary-operation + normalized = typehints.normalize(pipe_union) + self.assertIsInstance(normalized, typehints.UnionConstraint) + pipe_union_2 = int | float # pylint: disable=unsupported-binary-operation + normalized_2 = typehints.normalize(pipe_union_2) + self.assertIsInstance(normalized_2, typehints.UnionConstraint) + class TestNonBuiltInGenerics(unittest.TestCase): def test_no_error_thrown(self): diff --git a/sdks/python/apache_beam/yaml/examples/testing/examples_test.py b/sdks/python/apache_beam/yaml/examples/testing/examples_test.py index 4f0516a1ea93..15cf46218e8e 100644 --- a/sdks/python/apache_beam/yaml/examples/testing/examples_test.py +++ b/sdks/python/apache_beam/yaml/examples/testing/examples_test.py @@ -563,8 +563,11 @@ def _wordcount_minimal_test_preprocessor( return _wordcount_random_shuffler(test_spec, all_words, env) -@YamlExamplesTestSuite.register_test_preprocessor( - ['test_wordCountInclude_yaml', 'test_wordCountImport_yaml']) +@YamlExamplesTestSuite.register_test_preprocessor([ + 'test_wordCountInclude_yaml', + 'test_wordCountImport_yaml', + 'test_wordCountInheritance_yaml' +]) def _wordcount_jinja_test_preprocessor( test_spec: dict, expected: List[str], env: TestEnvironment): """ @@ -679,6 +682,7 @@ def _kafka_test_preprocessor( 'test_anomaly_scoring_yaml', 'test_wordCountInclude_yaml', 'test_wordCountImport_yaml', + 'test_wordCountInheritance_yaml', 'test_iceberg_to_alloydb_yaml' ]) def _io_write_test_preprocessor( @@ -1256,8 +1260,11 @@ def _batch_log_analysis_test_preprocessor( return test_spec -@YamlExamplesTestSuite.register_test_preprocessor( - ['test_wordCountInclude_yaml', 'test_wordCountImport_yaml']) +@YamlExamplesTestSuite.register_test_preprocessor([ + 'test_wordCountInclude_yaml', + 'test_wordCountImport_yaml', + 'test_wordCountInheritance_yaml' +]) def _jinja_preprocessor(raw_spec_string: str, test_name: str): """ Preprocessor for Jinja-based YAML tests. diff --git a/sdks/python/apache_beam/yaml/examples/testing/input_data.py b/sdks/python/apache_beam/yaml/examples/testing/input_data.py index fb468567355d..7fe9b5291e01 100644 --- a/sdks/python/apache_beam/yaml/examples/testing/input_data.py +++ b/sdks/python/apache_beam/yaml/examples/testing/input_data.py @@ -86,6 +86,11 @@ def word_count_jinja_template_data(test_name: str) -> list[str]: 'apache_beam/yaml/examples/transforms/jinja/' 'import/macros/wordCountMacros.yaml' ] + elif test_name == 'test_wordCountInheritance_yaml': + return [ + 'apache_beam/yaml/examples/transforms/jinja/' + 'inheritance/base/base_pipeline.yaml' + ] return [] diff --git a/sdks/python/apache_beam/yaml/examples/transforms/jinja/inheritance/README.md b/sdks/python/apache_beam/yaml/examples/transforms/jinja/inheritance/README.md new file mode 100644 index 000000000000..e22e54a56696 --- /dev/null +++ b/sdks/python/apache_beam/yaml/examples/transforms/jinja/inheritance/README.md @@ -0,0 +1,77 @@ + + +# Jinja Inheritance Example + +This folder contains an example of how to use Jinja2 inheritance in Beam YAML pipelines. + +## Files + +* **base/base_pipeline.yaml**: A complete WordCount pipeline (Read -> Split -> Explode -> Combine -> MapToFields -> Write). It defines a block `extra_steps` between `Explode` and `MapToFields` to allow child pipelines to inject additional transforms. +* **wordCountInheritance.yaml**: Extends `base/base_pipeline.yaml` and injects a `Combine` transform into the `extra_steps` block to combine words. + +## Running the Example + +To run the child pipeline (which includes the inherited base pipeline logic + the new filter): + +General setup: +```sh +export PIPELINE_FILE=apache_beam/yaml/examples/transforms/jinja/inheritance/wordCountInheritance.yaml +export KINGLEAR="gs://dataflow-samples/shakespeare/kinglear.txt" +export TEMP_LOCATION="gs://MY-BUCKET/wordCounts/" +export PROJECT="MY-PROJECT" +export REGION="MY-REGION" + +cd /beam/sdks/python +``` + +Multiline Run Example: +```sh +python -m apache_beam.yaml.main \ + --project=${PROJECT} \ + --region=${REGION} \ + --yaml_pipeline_file="${PIPELINE_FILE}" \ + --jinja_variables='{ + "readFromTextTransform": {"path": "'"${KINGLEAR}"'"}, + "mapToFieldsSplitConfig": { + "language": "python", + "fields": { + "value": "1" + } + }, + "explodeTransform": {"fields": "word"}, + "combineTransform": { + "group_by": "word", + "combine": {"value": "sum"} + }, + "mapToFieldsCountConfig": { + "language": "python", + "fields": {"output": "word + \" - \" + str(value)"} + }, + "writeToTextTransform": {"path": "'"${TEMP_LOCATION}"'"} + }' +``` + +Single Line Run Example: +```sh +python -m apache_beam.yaml.main --project=${PROJECT} --region=${REGION} \ +--yaml_pipeline_file="${PIPELINE_FILE}" --jinja_variables='{"readFromTextTransform": +{"path": "'"${KINGLEAR}"'"}, "mapToFieldsSplitConfig": {"language": "python", "fields":{"value":"1"}}, "explodeTransform":{"fields":"word"}, "combineTransform":{"group_by":"word", "combine":{"value":"sum"}}, "mapToFieldsCountConfig":{"language": "python", "fields":{"output":"word + \" - \" + str(value)"}}, "writeToTextTransform":{"path":"'"${TEMP_LOCATION}"'"}}' +``` + diff --git a/sdks/python/apache_beam/yaml/examples/transforms/jinja/inheritance/base/base_pipeline.yaml b/sdks/python/apache_beam/yaml/examples/transforms/jinja/inheritance/base/base_pipeline.yaml new file mode 100644 index 000000000000..209646b894a5 --- /dev/null +++ b/sdks/python/apache_beam/yaml/examples/transforms/jinja/inheritance/base/base_pipeline.yaml @@ -0,0 +1,55 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +pipeline: + type: chain + transforms: + - type: ReadFromText + config: + path: {{readFromTextTransform.path}} + + - type: MapToFields + name: Split words + config: + language: python + fields: + word: + callable: |- + import re + def my_mapping(row): + return re.findall(r'[A-Za-z\']+', row.line.lower()) + value: {{mapToFieldsSplitConfig.fields.value}} + - type: Explode + config: + fields: + - {{explodeTransform.fields}} + + # Inheritance injection point: content added here by child pipelines will be executed + # after Explode and before MapToFields. +{% block extra_steps %} +{% endblock %} + + - type: MapToFields + name: Format output + config: + language: {{mapToFieldsCountConfig.language}} + fields: + output: {{mapToFieldsCountConfig.fields.output}} + - name: Write to GCS + type: WriteToText + config: + path: {{writeToTextTransform.path}} diff --git a/sdks/python/apache_beam/yaml/examples/transforms/jinja/inheritance/wordCountInheritance.yaml b/sdks/python/apache_beam/yaml/examples/transforms/jinja/inheritance/wordCountInheritance.yaml new file mode 100644 index 000000000000..ad9f44df7851 --- /dev/null +++ b/sdks/python/apache_beam/yaml/examples/transforms/jinja/inheritance/wordCountInheritance.yaml @@ -0,0 +1,40 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +{% extends "apache_beam/yaml/examples/transforms/jinja/inheritance/base/base_pipeline.yaml" %} + +{% block extra_steps %} + - name: Count words + type: Combine + config: + group_by: + - {{combineTransform.group_by}} + combine: + value: {{combineTransform.combine.value}} +{% endblock %} + +# Expected: +# Row(output='king - 311') +# Row(output='lear - 253') +# Row(output='dramatis - 1') +# Row(output='personae - 1') +# Row(output='of - 483') +# Row(output='britain - 2') +# Row(output='france - 32') +# Row(output='duke - 26') +# Row(output='burgundy - 20') +# Row(output='cornwall - 75') \ No newline at end of file diff --git a/sdks/python/build.gradle b/sdks/python/build.gradle index 970020da8605..6e0786d98553 100644 --- a/sdks/python/build.gradle +++ b/sdks/python/build.gradle @@ -200,6 +200,7 @@ platform_identifiers_map.each { platform, idsuffix -> } getVersionsAsList('python_versions').each { it -> def pyversion = it.replace('.', '') + def cibuildwheel_version = it == '3.10' ? '2.23.3' : '3.3.1' project.tasks.register("bdistPy${pyversion}${platform}") { description "Build a Python wheel distribution for Py${pyversion} ${platform}" @@ -220,7 +221,7 @@ platform_identifiers_map.each { platform, idsuffix -> args '-c', ". ${envdir}/bin/activate && " + // note: sync cibuildwheel version with GitHub Action // .github/workflows/build_wheel.yml:build_wheels "Install cibuildwheel" step - "pip install cibuildwheel==2.23.3 setuptools && " + + "pip install cibuildwheel==${cibuildwheel_version} setuptools && " + "cibuildwheel --print-build-identifiers --platform ${platform} --archs ${archs} && " + "cibuildwheel --output-dir ${buildDir} --platform ${platform} --archs ${archs} " }