使用数据流Kafka到bigquery模板时出错



我正在使用数据流kafka-to-bigquery模板。启动数据流作业后,它会在队列中停留一段时间,然后失败,并出现以下错误:

Error occurred in the launcher container: Template launch failed. See console logs.

当查看日志时,我看到以下堆栈跟踪:

at org.apache.beam.runners.dataflow.DataflowRunner.run(DataflowRunner.java:192) 
at org.apache.beam.sdk.Pipeline.run(Pipeline.java:317) 
at org.apache.beam.sdk.Pipeline.run(Pipeline.java:303) 
at com.google.cloud.teleport.v2.templates.KafkaToBigQuery.run(KafkaToBigQuery.java:343) 
at com.google.cloud.teleport.v2.templates.KafkaToBigQuery.main(KafkaToBigQuery.java:222) 
Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata –

在启动工作时,我提供了以下参数:

  1. 卡夫卡主题名称
  2. 引导服务器名称
  3. bigquery主题名称
  4. SA电子邮件
  5. 区域

我的卡夫卡主题仅包含以下信息:你好

kafka安装在gcp实例中,该实例与数据流工作者位于同一区域和子网中。

在这里添加这个作为后代的答案:

"获取主题元数据时超时";表示Kafka客户端无法连接到代理以获取元数据。这可能是由于各种原因造成的,例如工作虚拟机无法与代理进行通信(您是通过公共或私人ip进行通信?如果使用公共ip,请检查传入的防火墙设置(。也可能是由于端口不正确或代理需要SSL连接。一种确认方法是将Kafka客户端安装在与Dataflow Worker位于同一子网的GCE VM上,然后验证Kafka客户机是否可以连接到Kafka代理。

请参阅[1]为Kafka客户端配置ssl设置(您可以在GCE实例上使用cli进行测试(。管理代理的团队可以告诉您他们是否需要SSL连接。

[1]https://docs.confluent.io/platform/current/kafka/authentication_ssl.html#clients

嘿,谢谢大家的帮助,我正试图用内部ip访问kafka。当我把它放到公共ip上时,它就起了作用。事实上,我在同一个子网中运行kafka机器和工人。所以它也应该与内部ip一起工作。。。我现在正在检查

最新更新