CODEWITHSUNDEEP
pythonapache-kafkagoogle-cloud-dataflowapache-beam
Christian
Christian

Reputation: 79

ReadFromKafka with python in apache-beam Unsupported signal: 2

I´ve been strugglin making this work, I know this is a cross-language transform and all of that and I installed the Java jdk on my pc (when I write java -version on cmd I get correct information and all of that) but when I am trying to make a simple pipeline work:

import apache_beam as beam
from apache_beam.io.external.kafka import ReadFromKafka
from apache_beam.options.pipeline_options import PipelineOptions
import os

os.environ['GOOGLE_APPLICATION_CREDENTIALS']='credentialsOld.json'

  
def main():
print('======================================================')
beam_options = PipelineOptions(runner='DataflowRunner',temp_location=temp_location,staging_location=staging_location,project=project,experiments=['use_runner_v2'],streaming=True)

with beam.Pipeline(options=beam_options) as p:
    msgs = p | 'ReadKafka' >> ReadFromKafka(consumer_config={'bootstrap.servers':'xxxxx-xxxxx...','group_id':'testAB'},topics=['users'])
    msgs | beam.FlatMap(print)
    
if __name__ == '__main__':
    
  main()

I get this error: ValueError: Unsupported signal: 2

I have tried adding the parameter expansion_service= 'beam:external:java:kafka:read:v1' to the ReadFromKafka but then I get:

status = StatusCode.UNAVAILABLE

details = "DNS resolution failed for beam:external:java:kafka:read:v1: UNKNOWN: OS Error"

Im working on a venv python enviroment if this info can be usefull and my kafka cluster is on confluent cloud.

Im also getting this runtime error: RuntimeError: java.lang.RuntimeException: Failed to get dependencies of beam:transform:org.apache.beam:kafka_read_without_metadata:v1 from spec urn: "beam:transform:org.apache.beam:kafka_read_without_metadata:v1"

EDIT: Im getting the bootstrap server option from here enter image description here

Upvotes: 1

Views: 1210

Answers (1)

Christian
Christian

Reputation: 79

My mistake was that I was skippig the step where I have to start a expansion_service, I did that with this command java -jar beam-sdks-java-io-expansion-service-2.37.0.jar 8088 --javaClassLookupAllowlistFile='*' after downloading the beam-sdks-java-io-expansion-service-2.37.0.jar from https://mvnrepository.com/artifact/org.apache.beam/beam-sdks-java-io-expansion-service/2.36.0 and then specifying the port in expansion_service='localhost:8088'

Then I had two minor mistakes one was that I was using the JDK 18 and I think it wasnt compatible https://beam.apache.org/get-started/quickstart-java/ so I switched to JDK 17 and used python 3.8 instead of python 3.10

Upvotes: 1

Related Questions