在Apache Flink中使用正则表达式订阅Kafka主题
Apache Flink是一个开源的流处理框架，它提供了强大的流处理和批处理功能。在Apache Flink中使用正则表达式订阅Kafka主题，可以通过以下步骤实现：
- 导入必要的依赖：在Flink项目的pom.xml文件中添加Kafka和Flink Kafka连接器的依赖。
- 创建Flink Kafka Consumer：使用Flink Kafka连接器提供的KafkaConsumer类创建一个消费者实例。可以通过设置Kafka主题、Kafka集群地址、反序列化器等参数来配置消费者。
- 使用正则表达式订阅主题：在创建消费者实例后，可以使用正则表达式来订阅符合特定模式的Kafka主题。例如，可以使用"topic.*"来订阅所有以"topic"开头的主题。
- 处理接收到的消息：通过实现Flink的ProcessFunction或使用Flink提供的其他操作符对接收到的Kafka消息进行处理。可以根据业务需求进行数据转换、过滤、聚合等操作。
- 启动Flink作业：将处理逻辑封装为Flink作业，并通过Flink的执行环境（如StreamExecutionEnvironment）启动作业。
Apache Flink的优势在于其高吞吐量、低延迟和容错性。它适用于处理实时数据流，并支持事件时间和处理时间的语义。Flink还提供了丰富的状态管理和容错机制，可以保证数据处理的准确性和可靠性。
推荐的腾讯云相关产品是腾讯云流计算Oceanus，它是一种基于Apache Flink的流式计算服务。腾讯云Oceanus提供了高可用、高性能的流式计算能力，可以与腾讯云的其他产品（如消息队列CMQ、对象存储COS等）进行集成，实现端到端的数据处理和分析。更多关于腾讯云Oceanus的信息可以参考腾讯云Oceanus产品介绍。
相关·内容
2回答
我正在尝试使用AVRO 1.0.3从Kafka主题读取Flink数据。
我只知道这个特别的Kafka主题是有AVRO编码的消息，我有AVRO模式文件。
我的Flink代码：
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
Properties properties = new Properties();
浏览 3提问于2016-08-02得票数 3
回答已采纳
我正在集成Flink 1.8流和kafka 2.x(kafka_2.12-2.2.0，kafka_2.12-2.3.0)，并在启动flink作业几秒钟后遇到异常：
org.apache.kafka.common.protocol.types.SchemaException: Error computing size for field 'topics': Error computing size for field 'name': Missing value for field 'name' which has no default va
浏览 239提问于2019-07-23得票数 0
Flink版本1.9.0 Scala版本2.11.12 Kafka集群2.3.0版 我正在尝试将我创建的flink作业连接到有3个分区的kafka集群。我已经在我的本地主机上运行的kafka集群主题上测试了我的工作，它有一个分区，它可以对本地kafka进行读写。当我尝试连接到具有多个分区的主题时，我得到以下错误(topicName是我试图使用的主题的名称。奇怪的是，当我试图生成一个多分区的主题时，我没有任何问题。 java.lang.RuntimeException: topicName
at org.apache.flink.streaming.connectors.kafka.i
浏览 87提问于2019-12-06得票数 0
回答已采纳
1回答
我的目标是建立一个高吞吐量集群，使用Kafka作为源，使用Flink作为流处理引擎。这就是我所做的。
我已经在主服务器和工作人员上设置了一个2节点集群，配置如下。
主flink-flink.Master
jobmanager.rpc.address: <MASTER_IP_ADDR> #localhost
jobmanager.rpc.port: 6123
jobmanager.heap.mb: 256
taskmanager.heap.mb: 512
taskmanager.numberOfTaskSlots: 50
parallelism.default: 100
W
浏览 7提问于2015-08-28得票数 9
回答已采纳
我想将流SQL放在Kafka中，以供Flink for CEP使用。这是一种好方法吗？
我知道动态模式定义在Flink上是不允许的，我需要应用它们可以在无界事件流上更改的规则。
举个例子；
有一个供用户定义其设备规则的UI。想象一下，有一个事物跟踪服务(如亚马逊网络服务中心)保持物理设备的状态，我也认为这是将每个设备的特定规则放入影子规则中的好方法，当影子参与者接收到传感器数据时，它可以发出数据，并将其规则添加到IoT作为规则引擎使用(通过kafka)。因此，我希望Flink执行传入的传感器数据(及其规则)，这些数据对于每个设备都可能是不同的。
浏览 7提问于2017-07-20得票数 1
2回答
Flink不对kafka进行抵消
、、、、
我有一个flink流媒体工作，是从卡夫卡读取数据，并只是记录它。我启用了检查点。
我看不出卡夫卡中的折中，相反，我在错误之下。
任何帮助都是徒劳无功的。
{$KAFKA_HOME/bin/kafka-consumer-groups.sh --new-consumer --bootstrap-server localhost:9092 --describe --group flink-consumer-group
Error while executing consumer group command Group flink-consumer-group with protocol type &
浏览 5提问于2017-02-06得票数 3
回答已采纳
我通过下面的说明创建了一个主题：
C:\kafka_2.12-0.10.2.1>.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic test < C:\User11\Desktop\Data.csv
然后我测试了这个主题，得到了正确的数据。之后，我想在Flink program.My程序中打印主题如下：
try{
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnviron
浏览 1提问于2018-11-05得票数 0
回答已采纳
目前，我有一个Flink集群，它希望用一个模式来使用Kafka主题，通过这种方式，我们不需要维护一个硬代码Kafka主题列表。
import java.util.regex.Pattern;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
...
private static final Pattern topicPattern = Pattern.compile("
浏览 1提问于2019-07-30得票数 1
回答已采纳
我已经创建了一个简单的Apache项目，它将从Kafka主题读取数据并将数据写入S3桶。当我运行这个项目时，我没有收到任何错误，它成功地读取了卡夫卡主题中的每条消息，但是没有任何东西写到我的S3桶中。没有错误，所以很难尝试调试正在发生的事情。下面是我的项目和配置。只有在我使用StreamExecutionEnviornment时才会发生这种情况。如果我尝试使用常规的批处理S3来生成到ExecutionEnviornment，它就能工作。
S3测试Java程序
public class S3Test {
public static void main(String[] args) throws
浏览 3提问于2017-01-04得票数 1
回答已采纳
使用flink版本1.13.1
我写了一个自定义的度量报告，但在我的flink中似乎不起作用。启动flink时，JobManager显示警告日志，如下所示：
2021-08-25 14:54:06,243 WARN org.apache.flink.runtime.metrics.ReporterSetup [] - The reporter factory (org.apache.flink.metrics.kafka.KafkaReporterFactory) could not be found for reporter kafka. Available f
浏览 10提问于2021-08-25得票数 2
回答已采纳
在Flink SQL上使用带有Kerberos身份验证的Kafka数据源，Flink的本地测试已经通过，但当我将任务推到纱线上时，错误消息提示无法找到Kafka的JAAS文件！
与Flink SQL相关的代码如下：
create table source_sensor(
id VARCHAR,
ts bigint,
vc double)
WITH (
'connector' = 'kafka',
'topic' = 'huangfu_0110'
浏览 23提问于2022-01-24得票数 1
我正在尝试使用Scala和Flink将消息发布到Kafka主题中。但是，当使用中提供的代码创建FlinkKafkaProducer对象时，它告诉我无法应用构造器。以下是代码示例：
val studentProducer = new FlinkKafkaProducer[String](
"my_topic", // target topic
new SimpleStringSchema(), // serialization schema
properties, // producer config
浏览 103提问于2021-05-24得票数 1
版本flink(1.11.3)，kafka(2.1.1)
我的flink数据芹菜是卡夫卡(来源) -> flink ->卡夫卡(接收器)。
当我第一次提交工作时，效果很好。但是，在作业经理或任务经理失败后，如果重新启动，则会出现异常。
o.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer 2020-12-31 10:35:35:23.831 objectOperator -> Sink: objectSink (1/1)警告 -遇到错误org.apache.kafka.common.errors.Invali
浏览 5提问于2020-12-30得票数 1
我正在编写一个如下所示的Flink Kafka集成程序，但是得到了kafka的超时错误：
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer010,
FlinkKafkaProducer010}
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import java.util.Properties
object St
浏览 6提问于2017-12-15得票数 0
我有一个Flink作业，它在本地运行得很好，但当我尝试在集群上运行flink作业时失败。当尝试通过'connector‘= ' Kafka’从kafka加载数据时发生错误。我使用Flink-Table API和confluent-avro格式从Kafka读取数据。 所以基本上我创建了一个从kafka主题中读取数据的表： val inputTableSQL =
s"""CREATE TABLE input_table (
| -- key of the topic
| key BYTES NOT
浏览 299提问于2021-10-22得票数 0
回答已采纳
我想在这里添加一些代码，并从Flink中标准输出protobuf数据。 我正在使用Flink的Apache Kafka连接器，以便将Flink连接到Kafka。 这是我的Flink的代码。 val env = StreamExecutionEnvironment.getExecutionEnvironment
val props = new Properties()
props.setProperty("bootstrap.servers", "localhost:9092").
val producer = FlinkKafkaProducer011(top
浏览 11提问于2020-08-13得票数 0
我有一个基于flink的流媒体应用程序，它使用apache kafka源和接收器。因为有时候我会在开发过程中的随机时间得到异常，我不知道它们是从哪里来的。 我使用mainRunner类在IntelliJ中运行应用程序，并通过kafka向它提供消息。有时第一条消息会触发错误，有时只有在几条消息之后才会发生。 它看起来是这样的： 16:31:01.935 ERROR o.a.k.c.producer.KafkaProducer - Interrupted while joining ioThread
java.lang.InterruptedException: null
at
浏览 111提问于2019-01-29得票数 1
回答已采纳
下面是SBT配置
// https://mvnrepository.com/artifact/org.apache.flink/flink-scala
libraryDependencies += "org.apache.flink" %% "flink-scala" % "1.9.1"
// https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-scala
libraryDependencies += "org.apache.flink" %%
浏览 1提问于2020-02-27得票数 1
回答已采纳
1回答
我使用Apache Flink将来自给定的Kafka主题集的更新传播到Elasticsearch集群中。
我面临的问题是，有时Elasticsearch集群会进化，而我必须通过(1)修改映射，当我不得不将Flink作业指向新的别名/索引时，(2)复制到data...and时，就会有大量的更新到旧的索引。
所以我想知道什么是最好的方法来解决这个问题。我可以休息，但如果可能的话，我想避免这样做。我试图使Flink作业减速或暂停(Kafka)输入源，直到迁移完成，但我没有找到任何端点。
Flink作业在应用程序模式下运行。
如果有人能够解释一下如何做到这一点:通过API或其他类似工具暂停/恢复作
浏览 5提问于2022-06-21得票数 0
更新
为了提供一些上下文，使用LZ4压缩启用了Kafka主题。
当识别压缩类型时，尝试将LZ4与运行时由于某种原因可以找到的KafkaLZ4BlockInputStream相关联。
@Override
public InputStream wrapForInput(ByteBuffer inputBuffer, byte messageVersion, BufferSupplier decompressionBufferSupplier) {
try {
return new KafkaLZ4BlockInputStream(inputBuffer, decompres
浏览 18提问于2021-12-09得票数 1
我正在使用flink和kafka，我得到了这个错误。
java.lang.NoSuchMethodError: scala.Predef$.ArrowAssoc(Ljava/lang/Object;)Ljava/lang/Object;
at kafka.consumer.FetchRequestAndResponseMetrics.<init>(FetchRequestAndResponseStats.scala:32)
at kafka.consumer.FetchRequestAndResponseStats.<init>(FetchRequest
浏览 2提问于2016-06-30得票数 4
我有一个flink应用程序，它从kafka中读取并将其吸收到kafka。
当我从Intellij运行应用程序时，没有问题，但是当我将ClassCastException提交给flink集群时，它会给出shadowJar。我能在找出我在这里做错了什么方面得到一些帮助吗？
异常跟踪：
Caused by: java.lang.ClassCastException: cannot assign instance of org.apache.kafka.clients.consumer.OffsetResetStrategy to field org.apache.flink.connector.ka
浏览 61提问于2022-05-16得票数 4
我是flink和Kafka的新手。我正在尝试使用Confluent Schema注册表对avro数据进行反序列化。我已经在ec2机器上安装了flink和Kafka。此外，在运行代码之前已经创建了"test“主题。 代码路径：https://gist.github.com/mandar2174/5dc13350b296abf127b92d0697c320f2 作为实现的一部分，代码执行以下操作： 1) Create a flink DataStream object using a list of user element. (User class is avro generated
浏览 43提问于2019-04-22得票数 1
我在试着干涉kafka和flink。其思想是使用kafka队列并使用flink转换数据。我正在遵循下面提到的示例
这些是我的依赖项。
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.3.2</version>
</dependency>
浏览 4提问于2017-12-12得票数 1
回答已采纳
嗨，我用下面的代码发布从flink到Kafka的消息。
final Properties props = new Properties();
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALU
浏览 14提问于2022-06-14得票数 0
回答已采纳
1回答
我尝试在Google上Shell中的ssh控制台中导入这个库：
import org.apache.flink.connector.kafka.source.KafkaSource
我以前安装过flink：
wget https://dlcdn.apache.org/flink/flink-1.14.4/flink-1.14.4-bin-scala_2.11.tgz
tar -xzf flink-1.14.4-bin-scala_2.11.tgz
但是，当我运行导入行时，我会得到以下错误：
<console>:81: error: object kafka is not a mem
浏览 12提问于2022-05-12得票数 0
我有一个关于Flink-Kafka的问题来源： flink应用程序从检查点恢复后启动，并且运行良好。 在运行过程中，有几个Kafka分区被添加到Kafka主题中，运行中的flink应用程序是否会知道这些添加的分区，并在不需要手动操作的情况下读取它们？或者我必须重新启动应用程序，让flink在启动时知道这些分区？ 如果添加分区不需要手动操作，你能给我指出Flink处理Kafka分区更改的代码吗？我在代码中找不到逻辑。 谢谢!
浏览 23提问于2019-10-07得票数 1
。 嗨, 使用Apache Flink 1.8。我有一个来自Kafka的JSON记录流，并对它们进行过滤，一切都很好。 现在，我想用一个数据库表中的查找值来丰富Kafka中的数据。 这是否只是创建两个流，加载第二个流中的表，然后连接数据的情况？ 数据库表确实会更新，但并不频繁，我希望避免在通过流的每条记录上查找DB。
浏览 24提问于2019-04-16得票数 5
回答已采纳
我有一个Flink会话集群(作业管理器+任务管理器)，版本为1.11.1，配置了log4j-控制台属性，包括Kafka appender。另外，在作业管理器和任务管理器中，我都支持flink-s3-fs-hadoop内置插件。
我将kafka客户端jar添加到flink/lib目录中，这是容器运行所必需的。但是，在实例化S3插件(并初始化记录器)时，我仍然会得到以下类加载错误。
由: org.apache.kafka.common.config.ConfigException:无效值org.apache.kafka.common.serialization.ByteArraySerialize
浏览 2提问于2020-12-06得票数 3
当我试图从Kafka中读取时，我得到了以下错误
java.lang.IllegalArgumentException: Unknown compression type id: 5
at org.apache.kafka.common.record.CompressionType.forId(CompressionType.java:46)
at org.apache.kafka.common.record.Record.compressionType(Record.java:260)
at org.apache.kafka.common.record.LogEntry.isCompresse
浏览 0提问于2019-01-31得票数 1
3回答
我正在尝试阅读卡夫卡主题的数据，在Flink流中。我正在尝试运行下面的示例代码，作为APACHE 1.1.3文档中的示例: Apache连接器，
import java.util.Properties;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;
import org.apache.flink.streaming.api.datastrea
浏览 4提问于2017-01-09得票数 2
回答已采纳
Flink-kafka-consumer有两种类型的消费者，例如： 这两个消费者层次结构都扩展了相同的FlinkKafkaConsumerBase类。我想知道维护编号类背后的设计决策是什么？我们应该在什么时候使用其中一个呢？ 我注意到带有编号后缀的类有一些更多的功能(例如，分级限制)。 https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaCons
浏览 6提问于2020-09-14得票数 0
在将消息发布到flink应用程序的kafka主题时，会出现此错误。相同的代码在我们的测试环境中工作，具有类似的生产者配置，但在生产环境中失败。我找不到这个问题的原因。
2022-04-15 16:51:37,228 thread="Sink: spend-limit-publisher-sink (2/2)#15" level=WARN logger=o.a.f.r.t.Task - Sink: spend-limit-publisher-sink (2/2)#15 (d7b4646c840c3882bb784125393b484a) switched from RUNNIN
浏览 1提问于2022-04-15得票数 0
我有apache访问日志文件，我想在cassandra表中存储每个页面的访问计数(总数/每天/每小时)。
我正在尝试使用kafka连接从日志文件流到kafka主题。为了在Cassandra中增加指标计数器，我可以再次使用Kafka Connect吗？否则，这里应该使用哪些其他工具，例如kafka streams，spark，flink，kafka connect等？
浏览 0提问于2017-10-11得票数 0
我有一个带有Yarn的Flink集群，使用flink-quickstart-java原型构建一个演示项目。在使用'mvn干净包-Pbuild‘命令构建了一个胖jar之后，并使用'flink run -m丝- -yn 2./flink-快照-1.0.jar’提交程序后，程序会抛出以下异常：
org/apache/kafka/common/serialization/ByteArrayDeserializer : org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09.setDeserializer(Fli
浏览 1提问于2019-08-05得票数 3
有人能帮我解决以下错误吗？我认为版本有一些问题，但我不知道如何修复它。我试图运行Apache，我使用Kafka主题作为输入，在尝试运行Flink时，我得到了以下错误：
错误：
线程"main“中的异常:org/apache/flink/util/ org.apache.flink.cep.pattern.Pattern.where(Pattern.java:130) at hes.cs63.CEPMonitor.CEPMonitoring.main(CEPMonitoring.java:55)，由: java.lang.ClassNotFoundException: org.apac
浏览 3提问于2017-11-20得票数 1
1回答
我有一份Flink的工作，把数据写进卡夫卡。Kafka主题的最大消息大小设置为5MB，所以如果我尝试编写任何大于5MB的记录，它会抛出以下异常并将作业降低。
java.lang.Exception: Failed to send data to Kafka: The request included a message larger than the max message size the server will accept.
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerBase.checkErroneo
浏览 1提问于2018-09-13得票数 1
回答已采纳
在使用Spark Structured Streaming和Kinesis Streams时，我在重新处理流中积累的数据时遇到了读取不平衡的情况(而不是从最新数据读取)。
下图显示了作为流的一部分的运动碎片的读取速度的差异。
这使得spark作业丢弃了许多事件，因为事件时间非常不同的事件会被混淆，而那些被认为是旧的事件会被丢弃。
最近，一位团队成员建议改用Kafka。我对Apache Kafka解决这个问题持怀疑态度，因为AFAIK修复我上面描述的读取不平衡的唯一方法是在消费者级别引入某种协调。这就是Apache Flink的kinesis连接器在重新处理Kinesis stream
浏览 1提问于2021-07-04得票数 1
因此，我正在尝试在我的Flink Kafka流媒体作业中启用EXACTLY_ONCE语义以及检查点。 然而，我没有让它工作，所以我尝试从Github：https://github.com/apache/flink/blob/c025407e8a11dff344b587324ed73bdba2024dff/flink-end-to-end-tests/flink-streaming-kafka-test/src/main/java/org/apache/flink/streaming/kafka/test/KafkaExample.java下载测试样例代码 因此，运行此程序可以很好地工作。然而
浏览 73提问于2020-06-19得票数 8
在深入研究了许多SO帖子，甚至JIRA问题之后，我不知道该去哪里找了。Flink中的每个检查点都由于超时而失败，在作业的异常部分中，它会显示以下错误，但是作业本身并没有失败：
org.apache.kafka.common.errors.TimeoutException: Timeout expired after 600000milliseconds while awaiting InitProducerId
当禁用检查点时，有关Kafka的一切都如预期的那样工作，所以我的假设是，它可能与等待Kafka提交以便被确认的检查点有关(Semantic设置为EXACTLY_ONCE)。我记得读过关
浏览 19提问于2020-09-24得票数 2
回答已采纳
在SCALA代码中使用Kafka/FLink命令行-“命令”
当前，当我向Flink提交作业时，它通过命令行。与创建和删除卡夫卡主题相同。
是否有一个Scala来控制命令行接口命令的执行.因此，根据应用程序中的条件，我可以向flink提交一个作业，并从代码中创建/删除kafka。
我想在Scala代码中使用的命令
Starting Flink Server
bin/start-local.sh
Stopping Flink Server
bin/stop-local.sh
Submitting Job Flink
bin/flink run -c pathToMai
浏览 7提问于2017-10-09得票数 0
回答已采纳
1回答
我正在尝试基于flink:1.4.1-hadoop27-scala_2.11-alpine映像将Flink作业部署到集群中。这项工作是使用卡夫卡连接器源(flink-连接器- Kafka -0.11)，我正在尝试分配时间戳和水印。我的代码非常类似于中的Scala示例。但有了FlinkKafkaConsumer011
val myConsumer = new FlinkKafkaConsumer08[String]("topic", new SimpleStringSchema(), properties)
myConsumer.assignTimestampsAndWaterm
浏览 0提问于2018-03-05得票数 2
回答已采纳
我正在尝试连接到本地机器上的Kafka (2.1)，并在随Flink (1.7.2)附带的shell中阅读。
我正在做的事情是：
:require flink-connector-kafka_2.11-1.7.1.jar
:require flink-connector-kafka-base_2.11-1.7.1.jar
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.streaming.util.serialization.SimpleStringSc
浏览 0提问于2019-03-11得票数 1
回答已采纳
尝试将Flink 1.9 SQL-Client与Kafka一起使用，但未成功。在弄清楚所需的jar文件并将它们复制到lib目录后，当执行SELECT * FROM table-name时，我得到了以下运行时异常： Flink SQL> select * from default_catalog.default_database.member_customer_newsletters ;
[ERROR] Could not execute SQL statement. Reason:
java.lang.ClassNotFoundException: org.apache.kafka.c
浏览 231提问于2019-09-20得票数 0
src代码： /*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version
浏览 29提问于2020-01-19得票数 0
我正在使用Flink Connector Kafka 1.8.0。(取决于kafka-clients 2.0.1) https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka_2.12/1.8.0 使用gradle： // https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka
compile group: 'org.apache.flink', name: 'flink-connector
浏览 23提问于2019-12-26得票数 0
我想设置Flink，这样它就可以将数据流从Apache转换并重定向到MongoDB。出于测试目的，我构建在flink流-Connectors.kafka示例()之上。
卡夫卡流是正确红色的链接，我可以映射他们等，但问题发生时，我想保存每一个接收和转换的信息到MongoDB。我发现的关于MongoDB集成的唯一例子是来自github的flink-mongodb测试。不幸的是，它使用的是静态数据源(数据库)，而不是数据流。
我认为应该有一些DataStream.addSink实现的MongoDB，但显然没有。
实现这一目标的最佳途径是什么？我是否需要编写自定义接收器函数，还是可能遗漏了什么？也许应
浏览 3提问于2016-02-02得票数 11
回答已采纳
我正在尝试读取Kafka主题中的记录，该主题是由kafka connect jdbc源连接器生成的，以下是连接器配置： "name": "customers",
"config": {
"poll.interval.ms": "3000",
"table.poll.interval.ms": "10000",
"errors.log.enable": "false",
"errors.log.include.m
浏览 670提问于2021-02-01得票数 0
回答已采纳
需要帮助！谁能给我指引一条正确的路？
以下是我的代码和日志的片段。
DataStream<ObjectNode> stream = env.addSource(KafkaConsumer.getKafkaConsumer());
DataStream<MyDataObject> dataStream = stream.flatMap(new DataTransformation());
我使用flatMapFunction来处理我的输入对象并获得多个对象。
以下是stackTrace：
java.lang.RuntimeException: Buffer pool is
浏览 8提问于2018-01-16得票数 4
我尝试使用sasl机制(OAUTHBEARER)配置kafka身份验证(使用flink 1.9.2，kafka-client 2.2.0)。
当使用Flink与SASL身份验证时，我得到了下面的异常。卡夫卡是在一个厚厚的罐子与应用程序。
在远程调试之后，我发现我的回调处理程序有一个ChildFirstClassloader，并且ChildFirstClassloader属于另一个ChildFirstClassloader，因此下面的测试失败(OAuthBearerSaslClientFactory)：
if (!(Objects.requireNonNull(callbackHandler)
浏览 6提问于2021-09-06得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
