flink1.9.1 支持一个 source 指定消费多个 topics么?

classic Classic list List threaded Threaded
3 messages Options
Reply | Threaded
Open this post in threaded view
|

flink1.9.1 支持一个 source 指定消费多个 topics么?

bradyMk
This post was updated on .
CONTENTS DELETED
The author has deleted this message.
Reply | Threaded
Open this post in threaded view
|

Re: flink1.9.1 支持一个 source 指定消费多个 topics么?

nobleyd
从1.12开始支持。之前只能通过多张表union成一个view方式实现。

bradyMk <[hidden email]> 于2020年12月18日周五 下午3:53写道:

> Hi,想请教一下大家:
>
>
> 最近通过flink_taskmanager_job_task_operator_KafkaConsumer_records_consumed_rate指标发现,
> flink某个任务消费一个topic A 竟然比消费topic A,B,C,D一起的指标要高,
> 也就是我四个topic每秒消费的数据竟然还没其中一个topic每秒消费的数据高,
> 所以想请问:flink1.9.1 支持一个 source 指定消费多个 topics么?
> 我的代码如下:
>     val A= params.getProperty("kafka.scene.data.topic")
>     val B= params.getProperty("kafka.scene.log.topic")
>     val C= params.getProperty("kafka.event.topic")
>     val D= params.getProperty("kafka.log.topic")
>     import scala.collection.JavaConverters._
>     val topics = List[String](sceneDataTopic, sceneLogTopic, eventTopic,
> sdkLog).asJava
>     env .addSource(new FlinkKafkaConsumer011(topics, new
> JSONKeyValueDeserializationSchema(false), kafkaPro))
>
>
>
>
> -----
> Best Wishes
> --
> Sent from: http://apache-flink.147419.n8.nabble.com/
>
Reply | Threaded
Open this post in threaded view
|

Re: flink1.9.1 支持一个 source 指定消费多个 topics么?

nobleyd
不好意思没看清楚,我说的是sql哈。datastream api的话我记得很早就支持了。

赵一旦 <[hidden email]> 于2020年12月18日周五 下午7:31写道:

> 从1.12开始支持。之前只能通过多张表union成一个view方式实现。
>
> bradyMk <[hidden email]> 于2020年12月18日周五 下午3:53写道:
>
>> Hi,想请教一下大家:
>>
>>
>> 最近通过flink_taskmanager_job_task_operator_KafkaConsumer_records_consumed_rate指标发现,
>> flink某个任务消费一个topic A 竟然比消费topic A,B,C,D一起的指标要高,
>> 也就是我四个topic每秒消费的数据竟然还没其中一个topic每秒消费的数据高,
>> 所以想请问:flink1.9.1 支持一个 source 指定消费多个 topics么?
>> 我的代码如下:
>>     val A= params.getProperty("kafka.scene.data.topic")
>>     val B= params.getProperty("kafka.scene.log.topic")
>>     val C= params.getProperty("kafka.event.topic")
>>     val D= params.getProperty("kafka.log.topic")
>>     import scala.collection.JavaConverters._
>>     val topics = List[String](sceneDataTopic, sceneLogTopic, eventTopic,
>> sdkLog).asJava
>>     env .addSource(new FlinkKafkaConsumer011(topics, new
>> JSONKeyValueDeserializationSchema(false), kafkaPro))
>>
>>
>>
>>
>> -----
>> Best Wishes
>> --
>> Sent from: http://apache-flink.147419.n8.nabble.com/
>>
>