[Python] Managed Transforms API

[ahmedabu98]

Adding Python API for Managed transforms, similar to the Java API.

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 71.14%. Comparing base (bb4ad99) to head (a3067dc).
Report is 100 commits behind head on master.

❗ Current head a3067dc differs from pull request most recent head 0296f60

Please upload reports for the commit 0296f60 to get more accurate results.

Additional details and impacted files

@@            Coverage Diff            @@
##             master   #31495   +/-   ##
=========================================
  Coverage     71.14%   71.14%           
  Complexity     3008     3008           
=========================================
  Files          1055     1055           
  Lines        133439   133439           
  Branches       3248     3248           
=========================================
  Hits          94929    94929           
  Misses        35382    35382           
  Partials       3128     3128           

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[ahmedabu98]

Successful Dataflow run (2024-07-15_13_54_11-2004515317250011573) with the following code:

import logging
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

_LOGGER = logging.getLogger(__name__)
_LOGGER.setLevel(logging.INFO)

options = PipelineOptions([
    "--runner=DataflowRunner",
    "--job_name=managed-iceberg-read-demo",
    "--project=apache-beam-testing",
    "--temp_location=gs://apache-beam-testing-ahmedabualsaud/tmp",
    "--region=us-central1",
    "--sdk_location=sdks/python/dist/apache_beam-2.59.0.dev0.tar.gz",
    "--sdk_harness_container_image_overrides=.*java.*,gcr.io/apache-beam-testing/beam-sdk/beam_java8_sdk:latest"
])

with beam.Pipeline(options=options) as p:
    p | beam.managed.Read(
        "iceberg",
        config={
            "table": "my_database.my_table",
            "catalog_name": "ahmed_catalog",
            "catalog_properties": {
                "catalog-impl": "org.apache.iceberg.hadoop.HadoopCatalog",
                "warehouse": "gs://ahmedabualsaud-apache-beam-testing"
            }}) | "Log rows" >> beam.Map(_LOGGER.info)

Note: I had to include the following dependencies (needed by Iceberg) in the expansion service jar:

runtimeOnly library.java.hadoop_client
runtimeOnly library.java.bigdataoss_gcsio
runtimeOnly library.java.bigdataoss_gcs_connector
runtimeOnly library.java.bigdataoss_util_hadoop

[tvalentyn]

left some comments but Cham would have more context.

[tvalentyn]

left some comments but Cham would be have more context.

[chamikaramj]

Thanks!

[ahmedabu98]

@chamikaramj the last comment was just addressed. It's been a while though, lmk if you'd like to take another quick look before I merge

[chamikaramj]

Thanks! LGTM.

[chamikaramj]

Probably this file should be named "managed_iceberg_it_test.py"

[ahmedabu98]

Fixes #31830

[tvalentyn]

Looks like we merged this with
beam_PreCommit_RAT (Run RAT PreCommit) failing, and it is still broken at master @ahmedabu98 - could you PTAL? Thanks!

[tvalentyn]

> Task :rat
Files with unapproved licenses:
 - /runner/_work/beam/beam/sdks/python/apache_beam/transforms/managed_iceberg_it_test.py

[ahmedabu98]

Ahh my apologies, I didn't pay attention and thought the failure was a flake

[ahmedabu98]

Opened #32785