[YAML] ReadFromKafka with yaml

[Yarden BenMoshe <ya...@gmail.com>]

Hi,

I am trying to consume a kafka topic using ReadFromKafka transform.

If i got it right, since ReadFromKafka is originally written in java, an
expansion service is needed and default env is set to DOCKER, and in
current implementation I can see that expansion service field is not
adjustable (im not able to pass it as part of the transform's config).
Is there currently a way to ReadFromKafka from a pipeline written with yaml
api? If so, an explanation would be much appreciated.

I saw there's some workaround suggested online of using Docker-in-Docker
but would prefer to avoid it.

Thanks
Yarden

Re: [YAML] ReadFromKafka with yaml

[Ferran Fernández Garrido <ff...@gmail.com>]

Hi Yarden,

If you are using Dataflow as a runner, you can already use
ReadFromKafka (introduced originally in version 2.52). Dataflow will
handle the expansion service automatically, so you don't have to do
anything.

If you want to run it locally for development purposes, you'll have to
build the Docker image. You can check out the project and run:

./gradlew :sdks:java:container:java8:docker
-Pdocker-repository-root=$DOCKER_ROOT -Pdocker-tag=latest (DOCKER ROOT
-> repo location)

Then, for instance, if you want to run your custom Docker image in
Dataflow, you could do this:

(Build the Python SDK -> python setup.py sdist to get
apache-beam-2.53.0.dev0.tar.gz)

You'll have to build the expansion service that Kafka uses (in case
you've changed something in the KafkaIO) : ./gradlew
:sdks:java:io:expansion-service:build

python3 -m apache_beam.yaml.main --runner=DataflowRunner
--project=project_id --region=region --temp_location=temp_location
--pipeline_spec_file=yaml_pipeline.yml
--staging_location=staging_location
--sdk_location="path/apache-beam-2.53.0.dev0.tar.gz"
--sdk_harness_container_image_overrides=".*java.*,$DOCKER_ROOT:latest"
--streaming

This is an example of how to read JSON events from Kafka in Beam YAML:

- type: ReadFromKafka
  config:
      topic: 'TOPIC_NAME'
      format: JSON
      bootstrap_servers: 'BOOTSTRAP_SERVERS'
      schema: 'JSON_SCHEMA'

Best,
Ferran

El mié, 10 ene 2024 a las 14:11, Yarden BenMoshe
(<ya...@gmail.com>) escribió:
>
> Hi,
>
> I am trying to consume a kafka topic using ReadFromKafka transform.
>
> If i got it right, since ReadFromKafka is originally written in java, an expansion service is needed and default env is set to DOCKER, and in current implementation I can see that expansion service field is not adjustable (im not able to pass it as part of the transform's config).
> Is there currently a way to ReadFromKafka from a pipeline written with yaml api? If so, an explanation would be much appreciated.
>
> I saw there's some workaround suggested online of using Docker-in-Docker but would prefer to avoid it.
>
> Thanks
> Yarden