我想启用Flink从Pulsar读取数据
Flink是一个开源的流式数据处理框架，主要用于实时大数据处理和分析。它支持高吞吐量、低延迟的数据流处理，并且具备容错机制，可以保证数据处理的可靠性和一致性。Flink提供了丰富的API和工具，可以进行复杂的数据转换、聚合、窗口计算、模式匹配等操作。
Pulsar是一个开源的分布式流式消息系统，具有高吞吐量和低延迟的特点。它支持多租户、持久化存储和灵活的消息传递模式，并且可以水平扩展以满足大规模数据处理需求。Pulsar提供了可靠的消息传递和多种消费模式，可以灵活地适应不同的应用场景。
要使用Flink从Pulsar读取数据，可以按照以下步骤进行配置和操作：
- 安装和配置Flink：首先需要安装和配置Flink，可以从官方网站下载并按照文档进行配置。
- 引入Pulsar连接器：Flink提供了针对Pulsar的连接器，可以通过Maven等方式将其引入到项目中。
- 创建Pulsar消费者：使用Flink提供的Pulsar连接器，可以创建一个Pulsar消费者，并指定要读取的主题和消费者的配置。
- 实现数据处理逻辑：在Flink中，可以使用各种API（如DataStream API、Table API或SQL API）来实现对读取到的Pulsar数据的处理逻辑，例如转换、聚合、筛选等操作。
- 提交和执行任务：将编写好的Flink程序打包成可执行文件，然后通过Flink的客户端命令提交任务并启动执行。
推荐的腾讯云相关产品和产品介绍链接地址：
- 腾讯云Flink产品介绍：https://cloud.tencent.com/product/flink
- 腾讯云消息队列Pulsar产品介绍：https://cloud.tencent.com/product/cmqp
请注意，本回答仅涉及Flink和Pulsar的基本概念和使用方法，实际应用中可能还需要根据具体需求进行更详细的配置和开发。
相关·内容
Flink数据管道从Apache分区主题中读取。我已经将PulsarSource订阅设置为SubscriptionType.Exclusive。当它被更改为SubscriptionType.Shared时，它期望为命名空间启用事务策略。然后在代理中启用事务管理器并启动它。但还是得到了这个例外
Caused by: java.io.IOException: org.apache.flink.util.FlinkRuntimeException: java.util.concurrent.ExecutionException: org.apache.pulsar.client.api.transa
浏览 16提问于2022-05-28得票数 0
Apache pulsar通过分层存储减载器提供了一个非常有趣的架构。 我想知道如何从另一个应用程序进行交互式查询？我的意思是直接查询键值系统/“数据库”，而不是使用pulsar-sql，后者在底层使用Presto。 在@sijieg中，在twitter上，has posted以下模式： ? 看起来我可以访问State-store (甚至是Segment-reader)并直接访问博彩公司中的数据(根据Metastore ?)我们如何访问这些状态存储/段阅读器并访问数据，就像Flink-Pulsar或Spark-Pulsar一样。
浏览 17提问于2020-04-07得票数 2
1回答
我是新来Pulsar的！ 现在，我正在尝试用Scala实现这些代码( https://flink.apache.org/2019/05/03/pulsar-flink.html )。但是，我找不到一些类(例如PulsarSourceBuild)。我如何在Scala中实现它们？我应该在哪里检查？
浏览 20提问于2020-07-28得票数 0
回答已采纳
我正试图从脉冲星的话题中吸取经验。下面可以看到我的客户端代码。但是，我面对的是NoClassDefFoundError。请告诉我如何解决这个问题。
脉冲星客户端版本: 2.10，脉冲星服务器版本: 2.10，安装类型:独立(非码头)，操作系统: Ubuntu 20.04.4 LTS
我的脉冲星客户端代码：
private static void consumeFromPulsarAsync() throws Exception {
logger.info("consumeFromPulsarAsync()");
PulsarClient client = P
浏览 3提问于2022-05-10得票数 1
我一直在尝试各种例子，但我需要了解如何将PyFlink连接到Pulsar。我有Pulsar 2.8.0，Flink 1.13.1和Scala 2.11。我只需要了解如何设置PyFlink的参数来连接到Pulsar上的主题。
请帮帮忙。
浏览 3提问于2021-06-25得票数 0
我正在使用Flink从Apache Pulsar读取数据。我在pulsar中有一个有8个分区的分区主题。我在这个主题中生成了1000条消息，分布在8个分区中。我的笔记本电脑中有8个核心，所以我有8个子任务(默认情况下，并行度=核心数量)。在执行Eclipse中的代码后，我打开了Flink-UI，我发现一些子任务没有接收到任何记录(空闲)。我期望所有8个子任务都被利用(我期望每个子任务都映射到我的主题中的一个分区)。
重新启动作业后，我发现有时利用了3个子任务，有时利用了4个任务，而其余的子任务保持空闲。
请您的支持来澄清这个场景。
另外，我如何知道子镜头之间是否存在混洗？
我的代码：
Cons
浏览 0提问于2020-01-23得票数 1
我正在使用从Flink中的脉冲星读取数据。当数据的格式是Protocol Buffer时，我遇到了困难。
在GitHub的首页，Pulsar-Flink使用的是SimpleStringSchema。然而，看起来它并没有正式遵守Protocol Buffer。有人知道如何处理数据格式吗？我应该如何定义模式？
StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();
Properties props = new Properties();
props.setProperty("
浏览 17提问于2020-07-30得票数 1
回答已采纳
我是个新手，我在做Flink和Pulsar的工作。我在Flink的SlidingProcessingTimeWindows上有一个关于计算器与脉冲星的区别数据的任务。
我的windowSize: 60和窗口幻灯片: 5s
我的数据使用者每秒钟从主题脉冲星(收到2条消息/1秒)：
00:
- a.example.com
- a.example-2.com
---
01:
- b.example.com
- a.example-2.com
---
02:
- c.example.com
- a.example-2.com
---
03:
浏览 6提问于2022-08-15得票数 0
我正在使用flink从Azure数据湖中读取数据。但flink无法找到Azure数据湖文件系统。我已经实现了spark来读取Azure Data Lake文件系统文件。
因此，我尝试从flink中触发火花，从Azure数据湖中读取数据，并执行一些逻辑操作，然后将列表返回给flink。所以flink可以使用这些数据？
是否可以从flink触发spark作业？或者如何配置flink来理解Azure数据湖文件系统。有人能在这方面给我指点一下吗？
浏览 2提问于2017-07-11得票数 0
当从IDE本地启动Flink Web接口时，我想访问它。
我需要这个，因为我想访问Flink的计数器(累加器)。
浏览 39提问于2016-02-01得票数 12
回答已采纳
1回答
我是Apache Pulsar的新手，但我正在尝试使用Pulsar-Functions来处理某个主题的现有消息。这些函数似乎仅用于流式传输，因此只有在新消息到达输入主题时才会触发它们。不过，我还是想确认一下。 谢谢!
浏览 12提问于2021-08-06得票数 0
我已经写了一个非常简单的Flink流作业，从卡夫卡使用FlinkKafkaConsumer082的数据。
protected DataStream<String> getKafkaStream(StreamExecutionEnvironment env, String topic) {
Properties result = new Properties();
result.put("bootstrap.servers", getBrokerUrl());
result.put("zookeeper.connect", ge
浏览 0提问于2015-11-03得票数 2
上下文
我有一个由编写的Flink作业。它正在消耗动态的源数据，并产生结果到动态。我想做一个本地测试，以确保Flink应用程序代码是正确的。因此，我用文件系统连接器模拟了源Kinesis和接收器Kinesis。然后在本地运行测试管道。虽然本地flink作业总是成功运行。但当我查看水槽文件的时候。接收器文件总是空的。在“Flink SQL Client”中运行代码时也是如此。
这是我的代码：
CREATE TABLE incoming_data (
requestId VARCHAR(4),
groupId VARCHAR(32),
userId
浏览 12提问于2021-11-15得票数 0
我想设置Flink，这样它就可以将数据流从Apache转换并重定向到MongoDB。出于测试目的，我构建在flink流-Connectors.kafka示例()之上。
卡夫卡流是正确红色的链接，我可以映射他们等，但问题发生时，我想保存每一个接收和转换的信息到MongoDB。我发现的关于MongoDB集成的唯一例子是来自github的flink-mongodb测试。不幸的是，它使用的是静态数据源(数据库)，而不是数据流。
我认为应该有一些DataStream.addSink实现的MongoDB，但显然没有。
实现这一目标的最佳途径是什么？我是否需要编写自定义接收器函数，还是可能遗漏了什么？也许应
浏览 3提问于2016-02-02得票数 11
回答已采纳
当我们在作为SQL Server数据库表的源的Flink SQL CLI中选择任何表时，为什么Flink作业在从数据库表中获取所有记录后“完成”？有没有可能让它“运行”，这样它就可以从数据库表中获取最新的数据？
浏览 4提问于2020-12-25得票数 1
我需要我的Flink批处理作业从我的Spring Boot项目的postgres数据库中读取数据。我在Flink中看到的选项仅用于读取csv文件。在这种情况下，我需要从我的表创建一个csv文件。有什么解决方案吗？
浏览 36提问于2020-07-23得票数 2
回答已采纳
我想使用Flink-HBase插件读出数据，然后作为Flink机器学习算法的输入，分别是SVM和MLR。现在，我首先将提取的数据写入一个临时文件，然后通过libSVM方法读取它，但我想应该有一种更复杂的方法。
您有代码片段或知道如何做吗？
浏览 4提问于2015-07-12得票数 3
回答已采纳
我使用flink从文件中连续读取数据，该文件定期附加数据。我尝试在flink.But中使用readFile方法，不知道如何在此方法的参数中提及FileInputFormat。我的文件格式是json。有人能帮我处理一下this?.Thanks吗？
浏览 0提问于2017-06-21得票数 1
我正在构建一个要求低于要求的应用程序，我刚刚开始使用flink。
摄入卡夫卡的数据，比如50个分区(输入速率- 100,000毫希/秒)
从Kafka读取数据并对每一数据进行实时处理(进行一些计算、与旧数据的比较等)
将输出存储在Cassandra上
我正在寻找一个实时流媒体平台，并发现Flink是一个非常适合实时和批处理。
您认为flink最适合我的用例吗?还是应该使用Storm、Spark流或任何其他流平台？
我是否需要在google数据流中编写数据管道来执行flink上的步骤序列，还是有其他方法来执行实时流的步骤序列？
假设我的每一次计算花费大约20毫秒，
浏览 2提问于2016-06-10得票数 3
我已经从Run Pulsar locally在Ubuntu18.04.2LTS上安装了apache pulsar standalone v2.3。它不需要任何身份验证就能正常工作。我尝试过使用security-tls-transport和security-tls-authentication配置tls证书。我已经按照链接中提到的修改了以下配置broker.conf，client.conf和standalone.conf。使用命令bin/pulsar standalone运行脉冲星，并使用下面的命令发送消息。 bin/pulsar-client produce my-topic --messa
浏览 162提问于2019-04-03得票数 1
我研究Apache Pulsar已经有一段时间了，我很高兴。但是，我刚开始使用身份验证/授权，并且在Java管理客户机中遇到了一个奇怪的错误。
虽然我可以执行所有类型的操作，如创建/删除租户、创建/删除名称空间、创建/删除分区主题、向租户添加管理员角色等，但尝试使用
org.apache.pulsar.client.admin.Namespaces.grantPermissionOnNamespace()
我得到了这个错误：
org.apache.pulsar.client.admin.PulsarAdminException$ServerSideErrorException: HTTP 50
浏览 1提问于2020-05-20得票数 0
1回答
Apache通过从检查点恢复作业来保证在故障和恢复时处理一次，检查点是分布式数据流和操作符状态的一致快照(Chandy-Lamport算法用于分布式快照)。此保证在故障转移时恰好是一次。
在正常的集群操作中，Flink如何保证只处理一次，例如给定从外部源读取的Flink源(例如Kafka)，Flink如何保证事件从源读取一次？事件源和Flink源之间是否存在某种类型的应用程序级别？另外，Flink如何保证事件准确地从上游运营商到下游运营商之间只发生一次？对于接收到的事件，这也需要任何形式的附加条件吗？
浏览 3提问于2020-09-30得票数 3
回答已采纳
我需要我的flink作业在指定的时间间隔内从数据库中提取记录，并在处理后将其归档。我实现了SourceFunction，从数据库中获取所需的记录，并添加了SourceFunction作为StreamExecutionEnvironment的源。如何指定StreamExecutionEnvironment需要每10分钟使用一次SourceFunction从数据库中提取记录？
SourceFunction：
public class MongoDBSourceFunction implements SourceFunction<List<Book>>{
publi
浏览 0提问于2018-10-23得票数 2
回答已采纳
我正在使用支持Debezium源连接器的Pulsar IO，以便从您的数据库获取流更改。 从文档中可以清楚地看到如何连接到MYSQL， $ bin/pulsar-admin source localrun --sourceConfigFile debezium-mysql-source-config.yaml 监视连接器的步骤 bin/pulsar-admin functions getstatus \
--tenant <tenant> \
--namespace <namespace> \
--name <connector-nam
浏览 10提问于2019-08-31得票数 0
您能帮助我如何使用windows docker.i安装本地独立的pulsar集群吗?我已经按照下面的options.but操作，我无法访问pulsar UI
8080端口已经分配给其他process.so，这里我使用的是8081端口。
选项1：
docker运行-it -p 6650:6650 -p 8081:8081 --挂载数据，source=pulsardata =/pulsar/ source=pulsarconf --挂载数据，目标=/pulsar/conf apachepulsar/pulsar:2.5.2 bin/pulsar standalone
选项2: docker run
浏览 7提问于2020-06-08得票数 1
下面是我将脉冲星作为独立版本运行在坞-组合中的配置：
pulsar:
image: apachepulsar/pulsar:2.6.0
ports:
- 8080:8080
- 6650:6650
environment:
PULSAR_MEM: " -Xms512m -Xmx512m -XX:MaxDirectMemorySize=1g"
PULSAR_PREFIX_webSocketSessionIdleTimeoutMillis: 36000000
command: bash -c &
浏览 8提问于2022-08-18得票数 0
回答已采纳
1回答
如何并行读取mysql到flink的数据?我想要构建一个sourceFunction，在parallel.How中每隔一段时间从mysql中连续读取数据，我能实现吗？
浏览 12提问于2017-09-27得票数 2
我想让我的Flink SQL Client从Kafka中读取数据
卡夫卡产生的数据是Timestamp格式的，
但是默认的Flink SQL客户端解析是DateTime样式，
如何强制Flink SQL切换到Timestamp风格？
谢谢你的帮助~！
我的.yaml文件是
浏览 1提问于2020-12-23得票数 0
我在一个关系数据库中有一些批处理的数据，我想使用Flink将它们推送到消息总线上。由于Flink同时支持批处理和流模式，因此它看起来非常适合。话虽如此，我不知道这项任务是属于StreamingJob还是BatchJob，以及如何将两者联系起来。此任务是否更适合FlinkSQL环境？ 这个是可能的吗？我需要注意什么？
浏览 25提问于2021-04-28得票数 0
回答已采纳
我使用helm部署了脉冲星，没有激活tls。现在我想使用nginx入口暴露它，第一次不使用tls。它不工作，我在客户端得到了这个： Got exception TooLongFrameException : Adjusted frame length exceeds 5253120: 1213486164 - discarded 是否缺少特定的注解？ 第二次，我想使用ssl证书以安全的方式公开它。入口是否使用pulsar+ssl协议？我还需要特定的注解吗？ apiVersion: extensions/v1beta1
kind: Ingress
metadata:
name: pulsa
浏览 60提问于2020-07-23得票数 1
我正在开发一个应用程序，它将一些文件上传到s3桶，稍后，从s3桶中读取文件，并将其推送到数据库。
我使用Flink 1.4.2和fs.s3a API从s3桶读取和写入文件。
上传文件到s3桶没有任何问题，但是当我的应用程序的第二阶段--从s3读取这些上传的文件--启动时，我的应用程序会抛出错误
Caused by: java.io.InterruptedIOException: Reopen at position 0 on s3a://myfilepath/a/b/d/4: org.apache.flink.fs.s3hadoop.shaded.com.amazonaws.SdkClient
浏览 0提问于2019-06-21得票数 1
我使用Apache Pulsar Python客户端来连接我的微服务。我想创建Key_Shared消费者，如下所示：
import pulsar,_pulsar
client = pulsar.Client('pulsar://localhost:6650')
consumer = client.subscribe('my-topic', 'consume-id',
consumer_type=_pulsar.ConsumerType.KeyShared)
但我得到了以下错误：
Attr
浏览 13提问于2020-06-05得票数 1
我对Pulsar还不熟悉，我只是在探索一个新项目的功能。我正在尝试一个非常基本的例子，根据模式从生产者发送数据。为了给出一些背景，我的想法是将数据从apache发送到Clickhouse数据库。我已经完成了接收器连接器的设置，并使用下面的命令验证了相同的设置。
bin/pulsar-admin接收器状态--租户公共名称空间默认名称jdbc-clickhouse-接收器
bin/pulsar-admin接收器列表--租户公共名称空间默认输出：“jdbc-clickhouse-接收器”
所以我在Clickhouse DB中创建了一个表。我希望将数据发送到一个应该保存在数据库中的主题。在这样做的过程
浏览 8提问于2022-02-11得票数 3
我试图从cassandra表中提取数据作为数据集，但我遇到了两个问题。
首先，cassandraInputFormat只返回一个元组，我更希望没有tuple12，只需要使用pojo来定义它期望返回的内容。因此，我不知道这是否只是我必须接受的东西，是否有一种方法可以使用pojo代替cassandraConnector ()，或者使用cassandraInputFormat不是获取数据的最佳方式。
另一个问题是，即使我从cassandraInputFormat中提取的数据(不管是元组还是非元组)，我也不知道如何将它设置为数据源。对于文件、csv和HDFS，有很多方法()，但没有一个方法显式地用于c
浏览 4提问于2017-05-11得票数 0
回答已采纳
这样做的目的是从用户获取两个文本输入，将它们组合在一起，然后打印元素。我认为我走在正确的道路上，但是当我运行这段代码时，我得到了一个包含完全没有意义的元素的巨大的表。这是我目前拥有的代码：
defining the read function
def read(pulsar_name,signal_strength):
#opening and reading data from first line
pulsars = open(pulsar_name,"r").read()
#opening and reading data from s
浏览 2提问于2018-02-23得票数 0
1回答
我有一个在生产环境中运行的Flink流作业，我需要对主转换代码进行更改。
生产中的代码实际上如下所示：
stream
.filter(inboundData -> inboundData.hasToBeFiltered())
.uid("filtered-data")
.keyBy(data -> data.getMyStringKey())
.process(doSomething())
.uid("processed-inbound-data-id");
我需要改变keyBy()操作符使用inboundData PO
浏览 0提问于2019-08-08得票数 1
我使用的是带有32 32执行程序的flink mysql连接器，有32个插槽的16 16vCPU。如果我运行一个具有并行性32 (作业并行性224)的作业，它使用10个MySQL表执行时态查找联接，那么在2-3次成功运行之后，它就会开始失败。
org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandle
浏览 17提问于2022-01-17得票数 0
我使用elasticsearch和kibana保存和查询我的数据。一些很好的特性，如RBAC，SIEM需要在elasticsearch上与kibana进行ssl通信。所以我启用了xpack.security.http.ssl.enabled和xpack.security.transport.ssl.enabled。因此，客户端对es的请求必须通过https。
我还有一个flink集群，将数据写入elasticsearch。flink & elasticsearch和kibana正在码头群上运行。flink无需使用es进行身份验证或在通信量之间进行加密。我认为flink可以通过http协
浏览 5提问于2020-07-27得票数 1
回答已采纳
我是一个非常新的数据流，我希望建立管道，将使用pubsub作为源。
我曾经在流水线上工作过，它以flink为流引擎，kafka为源，因为我们可以在flink中设置并行性，以便从kafka读取消息，这样消息处理可以并行进行，而不是顺序进行。
我想知道在pubsub->dataflow中是否也可以这样做，或者它只按顺序读取消息。
浏览 2提问于2019-03-26得票数 0
在上学习了Apache的初学者Java教程之后，我想尝试对自己的数据进行一些转换。但是，我在收集运行在网络中的服务器上的Microsoft数据库的输入时遇到了问题。
关于的部分中的示例包含一个类似于我所需要的部分，其中一个DataSet是使用env.createInput(.)构建的。用JDBCInputFormat。因此，我为Flink JDBC添加了Maven依赖项
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-jdbc_2.11</
浏览 3提问于2016-11-10得票数 1
回答已采纳
1回答
我正在通过WebSockets使用Apache。当我使用读取器从主题中获取数据时，如果我确认了这些消息，它们会被从主题中删除吗？我已经为该主题的名称空间设置了一个保留期如下：
pulsar-admin namespaces set-retention public/default \
--size 1T \
--time -1
当我第一次使用阅读器时，我的设置就能工作；当数据进入主题时，我就能够阅读它。但是为什么不能使用读取器再次访问这些消息。
网址是：
ws://someEndPoint:8080/ws/v2/reader/persistent/public/default/topicNa
浏览 1提问于2019-11-17得票数 0
1回答
我使用的是在从docker-flink安装的ECS中运行的Flink。我已经通过flink-conf.yaml中的state.checkpoints.dir到S3启用了外部检查点到AWS S3。 现在，根据Flink文档here，如果我们想在出现故障的情况下从检查点恢复，我们必须使用bin/flink run -s :checkpointMetaDataPath [:runArgs]，但我使用FLINK_HOME/bin standalone-job.sh start-foreground。所以我不能弄清楚我的Flink作业在失败的情况下如何从外部化的检查点恢复。 我们真的需要一些从检查点恢
浏览 35提问于2020-04-03得票数 0
回答已采纳
我正在开发一个项目，该项目使用flink (Version - 1.4.2)将大量数据摄取到my图形数据库 (Janusgraph)。数据摄取有两个阶段，一个是顶点数据的摄取，另一个阶段是对图db的边缘数据的摄取。顶点数据摄取运行时没有任何问题，但是在边缘摄入期间，我遇到一个错误，即与任务管理器taskmanagerName失去了连接。来自flink-taskmanager-b6f46f6c8-fgtlw的详细错误跟踪附在下面：
2019-08-01 18:13:26,025 ERROR org.apache.flink.runtime.operators.BatchTask
- Err
浏览 1提问于2019-08-04得票数 0
我使用Elasticsearch连接器作为接收器将数据插入到Elasticsearch中(参见：https://ci.apache.org/projects/flink/flink-docs-release-1.7/dev/connectors/elasticsearch.html)。 但是，我没有找到任何连接器来从Elasticsearch获取数据作为源。 是否有任何连接器或示例使用Elasticsearch文档作为Flink pipline中的源？ 致以敬意， 阿里
浏览 24提问于2019-01-23得票数 1
我正在尝试使用Apache Pulsar的重复数据删除功能。 standalone.conf文件中设置了brokerDeduplicationEnabled=true，但是当我从生产者多次发送相同的消息时，我在消费者端得到了所有的消息，这是预期的行为吗？ 重复数据删除不就是AWS SQS中的content based deduplication吗？ 下面是我的生产者代码以供参考。 import pulsar
import json
client = pulsar.Client('pulsar://localhost:6650')
producer = cli
浏览 73提问于2021-08-18得票数 1
回答已采纳
1回答
我有一个关于在动态流中分片数据的问题。在将用户数据发送到我的动态流时，我希望使用一个随机分区键，以便碎片中的数据分布均匀。为了简化这个问题，我想通过在我的Flink应用程序中键入一个userId来聚合用户数据。
我的问题是:如果碎片是随机分区的，以便将一个userId的数据分散到多个Kinesis中，那么Flink能否处理多个碎片的读取，然后重新分发数据，以便将单个userId的所有数据都流到同一个聚合器任务中？或者，在Flink使用之前，我是否需要按用户id分割这个运动流？
浏览 3提问于2020-02-14得票数 0
回答已采纳
在我的问题中，我需要查询数据库并使用Flink中的Kafka数据流连接查询结果。目前，这是通过将查询结果存储在文件中，然后使用Flink的readFile功能来创建查询结果的DataStream来完成的。有什么更好的方法可以绕过编写文件和直接从查询结果创建DataStream的中间步骤呢？
我目前的理解是，我需要按照建议的SourceFunction编写一个自定义。这是正确的，也是唯一的办法，还是有其他选择？
是否有任何好的资源可用于编写自定义SoruceFunctions，或者应该只查看当前的实现以供参考，并根据自己的需要定制它们？
浏览 6提问于2022-03-17得票数 0
1回答
我正在尝试使用Flink 2.1.0从mysql日志表中读取流数据，但是，它只读取一次，然后它将停止该过程。我希望它继续读取，如果有传入的数据，并打印它。以下是我的代码
public class Database {
public static void main(String[] args) throws Exception {
// get the execution environment
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecution
浏览 72提问于2017-02-22得票数 2
我们有一个flink流应用程序，从卡夫卡读取消息。由于某些原因，我们不得不从kafka重置命令将kafka偏移重置为最新，因为有大量堆积。我们希望flink应用程序跳过所有这些消息，并从重置后出现的新消息开始。 问题是因为flink在内部管理它的偏移量，它不知道这个重置，它现在只从后向读取msg (重置前的偏移点)，现在也不能提交偏移量。因此，每次重启flink应用程序时，它都会再次从同一点读取。因此，我们在每次重启时都有重复的msg。 我知道我们不应该在flink kafka应用中手动重置偏移量。但是我们怎样才能从中恢复过来呢。 我已经尝试将auto.offset.config设置为最新，
浏览 62提问于2021-06-19得票数 4
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
