Flink批量数据处理
是一种流式计算框架，它能够高效地处理大规模数据集。以下是对Flink批量数据处理的完善且全面的答案：
概念：
Flink是一个开源的流式计算框架，它提供了批量数据处理和流式数据处理的能力。Flink的核心理念是将数据处理任务划分为有向无环图（DAG）的形式，通过对数据流的转换和操作，实现高效的数据处理。
分类：
Flink可以根据数据处理的方式进行分类，包括批量数据处理和流式数据处理。批量数据处理是指对一批数据进行处理，通常是在数据集已经完全生成之后进行处理。而流式数据处理是指对实时生成的数据流进行处理，数据可以无限延迟地产生。
优势：
- 高性能：Flink采用了基于内存的计算模型，能够在处理大规模数据时提供低延迟和高吞吐量的性能。
- 容错性：Flink具备故障自动恢复的能力，能够在节点故障时保证数据处理的连续性。
- 灵活性：Flink支持多种数据源和数据格式，可以与各种存储系统和消息队列集成，提供了灵活的数据处理能力。
- 可伸缩性：Flink能够根据数据量的增长自动扩展计算资源，保证在处理大规模数据时的高效性能。
应用场景：
Flink批量数据处理适用于以下场景：
- 批量数据分析：对大规模数据集进行离线分析，如数据清洗、数据挖掘、机器学习等。
- 批量数据转换：将数据从一种格式转换为另一种格式，如ETL（Extract-Transform-Load）过程中的数据转换。
- 批量数据计算：对大规模数据集进行复杂的计算，如图计算、统计分析等。
推荐的腾讯云相关产品：
腾讯云提供了一系列与Flink批量数据处理相关的产品和服务，包括：
- 云批量计算（BatchCompute）：提供了高性能的批量计算服务，可用于处理大规模数据集。
- 数据仓库（Data Warehouse）：提供了可扩展的数据存储和分析服务，适用于批量数据分析和计算。
- 弹性MapReduce（EMR）：提供了弹性的大数据处理服务，支持Flink等多种计算框架。
产品介绍链接地址：
- 云批量计算（BatchCompute）：https://cloud.tencent.com/product/bc
- 数据仓库（Data Warehouse）：https://cloud.tencent.com/product/dw
- 弹性MapReduce（EMR）：https://cloud.tencent.com/product/emr
相关·内容
1回答
我做了一个poc，在其中我使用火花流读取来自Kafka的数据。但是我们的组织要么使用Apache或者Kafka使用者来读取Apache中的数据作为一个标准过程。所以我需要用Kafka消费者或者Apache来代替Kafka流媒体。在我的应用程序用例中，我需要从kafka读取数据，过滤json数据，并在cassandra中放置字段，因此建议使用Kafka使用者，而不是flink/其他流，因为我实际上不需要对Kafka json数据进行任何处理。因此，我需要你的帮助来理解以下问题：
使用卡夫卡消费者，我能实现相同的连续数据读取，在火花流或flink的情况下吗？
考虑到我需要从kafka读取数
浏览 4提问于2017-03-26得票数 0
我想让我的Flink SQL Client从Kafka中读取数据
卡夫卡产生的数据是Timestamp格式的，
但是默认的Flink SQL客户端解析是DateTime样式，
如何强制Flink SQL切换到Timestamp风格？
谢谢你的帮助~！
我的.yaml文件是
浏览 1提问于2020-12-23得票数 0
我有flink应用程序，将在DC-1 (数据中心1)的节点上运行，我们计划使用HDFS或AMAZON-S3备份保存点和检查点状态。我的组织中对HDFS和S3的支持是，它不会将写入DC-1的数据复制到DC-2 (他们正在做这件事，但时间线很长)。考虑到这一点，有没有一种方法可以让flink本身以某种方式将flink检查点/保存点写入两个DC？谢谢
浏览 18提问于2021-11-22得票数 0
1回答
我希望有人能澄清TensorFlow与它的依赖关系(Beam、AirFlow、Flink等)
我指的是TFX的主页面：等等。
在示例中，我看到了三个变体： taxi_pipeline_flink.py、taxi_pipeline_kubeflow.py、taxi_pipeline_simple.py
梁的例子？
没有“梁”的例子，也很少描述它的用途。
假设即使没有安装气流，taxi_pipeline_simple.py也会运行，这是正确的吗？我认为没有，因为它使用"AirflowDAGRunner“。如果没有，那么你能运行TFX只有梁和它的转轮吗？如果是的话，为何没有这样的例子呢？
F
浏览 0提问于2019-05-17得票数 6
回答已采纳
当我使用pyflink hive sql read data insert into es时，抛出以下异常:环境: flink 1.11.2 flink-sql-connector-hive-3.1.2_2.11-1.11.2.jar hive 3.1.2 2020-12-17 21:10:24,398 WARN org.apache.flink.runtime.taskmanager.Task [] - Source: HiveTableSource(driver_id, driver_base_lc_p1, driver_90d_lc_p1, dr
浏览 338提问于2020-12-17得票数 0
下面是我观看的演示文稿中关于Flink优化器的幻灯片。对于Flink的优化器根据提供的数据集的基数来决定并行性的评论，我特别感到困惑。
我目前正在阅读Flink 1.4 (我正在使用的版本)文档，我似乎找不到任何关于Flink关于并行性的决定的文档。，我是否需要为Flink的优化器提供有关数据集的统计信息，以便利用这个特性？
在相关的说明中，我认为通过指定一个maxParallelism值，这可能使Flink能够动态地确定哪个级别的并行性适合于提供的数据集(如上面所详述的)。但是，我无法指定Flink 1.4文档所指定的最大并行性，这就是为什么我无法验证我的假设。对于某些上下文，我使
浏览 0提问于2019-05-20得票数 1
回答已采纳
我需要开始一个新的项目，我不知道是否火花或Flink会更好。目前，该项目需要微批处理，但以后可能也需要流事件处理。
假设火花是最好的，是否有缺点使用梁代替选择火花/Flink作为跑步者/引擎？
Beam是否会增加任何开销，或者在Spark/Flink中缺少某些API/功能？
浏览 2提问于2019-04-29得票数 1
回答已采纳
我很想知道用于外部数据访问的Asynchronous I/O Flink API是否也适用于Beam。我找不到任何具体的东西。我知道横梁使用Flink作为执行框架，所以异步I/O对横梁也是可能的，但我不确定API是否已经定义。
浏览 18提问于2021-10-06得票数 0
1回答
最近，我一直在比较火花和Flink的一个全新的项目。在这个项目中，流功能不是那么重要。对~(90) is数据进行批量分析是最重要的。稍后，我将在数据分析中应用ML和数据挖掘。
在搜索时，我发现很多文章，演示和视频声称Flink是下一代的分析解决方案。没有多少文章能为火花辩护。另一方面，星火是(或曾经？)非常流行，并广泛部署在非常大的生产系统。
我的问题是：对于我的用例来说，流并不重要，我是拥抱Flink还是从Spark2开始？
顺便说一下，我读过。这不能给我一个好的答案。
更新，2018年4月，：我们最终选择了火花。显然，除了表现之外，还有更多的问题要解决。Cloudera、Hortonwor
浏览 5提问于2017-05-19得票数 3
回答已采纳
如果我想在yarn上运行flink作业，命令是./bin/flink run -m yarn-cluster ./examples/batch/WordCount.jar 但是这个命令会运行一个默认的集群，它有2个任务管理器；如果我只提交了一个任务，为什么默认的任务管理器设置为2？ 什么时候我需要在一个任务中有多个任务管理器？
浏览 19提问于2020-07-02得票数 1
在需要时暂停或重新启动Flink作业的计算资源与持续运行Flink作业相比有什么好处？这是用于用例，新文件每天到达并在一小时内被摄入。这些工作现在一直在运行。但是想知道关于这种用例的任何最佳实践，这样就可以在向云迁移时对计算进行优化。
浏览 5提问于2021-06-18得票数 1
回答已采纳
从中，我知道循环可以用迭代运算符实现。由于Flink代码是延迟计算的，因此不可能使用while循环计算终止条件。
但是对于像initial.iterate(10){...}这样的批量迭代，我们可以使用while循环来达到同样的效果吗？
var dataset = ...
while (i < 10) {
dataset = step(dataset);
}
浏览 26提问于2017-12-20得票数 1
回答已采纳
我试图在Android应用程序中运行Apache Flink。我只想运行一个最小的工作示例，如下所示：
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
runFlinkExample();
}
private void runFlinkExample() {
StreamExecutionEnvironment env = Stre
浏览 0提问于2021-03-02得票数 0
在Flink中，一个作业可以有多个source->process->sink数据处理，例如 val env = StreamExecutionEnvironment.getExecutionEnvironment env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime) env.addSource(new Source1()).name("Source1").addSink(new Sink1()).name("Sink1").setParallelism(1) env.a
浏览 11提问于2020-08-30得票数 0
我是Flink的初学者。我可以实现Flink流。我还练习了关于KeyedStream的timeWindow。现在我想使用Flink关于DataStream的timeWindow来实现批处理操作。我尝试了一些代码，但它就是不能工作。我不知道怎样才能达到预期的效果。 val resultDataStream: DataStream[HBaseOperation] = env
.addSource(kafkaUserEventSource)
.assignTimestampsAndWatermarks(new Launcher.CustomWatermarkExtractor(Time.h
浏览 15提问于2019-09-09得票数 0
1回答
在jOOQ (v3.14.4)中使用批处理插入时，我在查看PostgreSQL (v12.6)日志时注意到了一些不一致性。在执行context.batch(<query>).bind(<1st record>).bind(<2nd record>)...bind(<nth record>).execute()时，日志显示记录实际上是逐个插入的，而不是一次性插入的。在执行过程中，context.insert(<fields>).values(<1st record>).values(<2nd record>)..
浏览 1提问于2021-02-21得票数 1
回答已采纳
1回答
我正在运行自己的GRPC服务器，收集来自各种数据源的事件。服务器是以Go开发的，所有事件源都以预定义的格式将事件作为protobuf消息发送。
我想要做的是在内存中处理所有这些事件。
我翻阅了Apache的文档，却找不到一个能做我想做的事情的样本。我不打算使用Kafka，Flink或任何其他流媒体平台，只需要处理内存中的消息并输出结果。
有人能告诉我一个正确的方法来开始编写一个简单的流处理应用程序吗？
浏览 1提问于2020-08-11得票数 0
回答已采纳
1回答
使用flink版本1.13.0 我在我的应用程序中使用广播状态，它将每2分钟加载大量数据( Map类型的大约500'000个数据) 我在web-ui中看到的拓扑图发现，每次广播源加载时，它都有50%-100%的背压，加入的过程有50%-100%的繁忙。我想知道在这段时间内进程是否会被阻塞和处理数据缓慢或停止处理数据？
浏览 27提问于2021-08-16得票数 0
回答已采纳
我们有一个使用NHibernate和Castle ActiveRecord的单一服务器/单一数据库应用程序，SQL Server是它的后盾。
作为处理的一部分，我需要写入多达120万行数据。使用ActiveRecord和NHibernate非常慢，需要改进-这可以通过"insert into .. select .. from ..“轻松实现。可能在一个存储的进程中。
拖我后腿的是ID生成。我们使用HiLo生成。如果我编写一个存储的proc来做我需要的事情，我可以从数据库表中获得下一个要使用的数字，执行插入操作，然后更新表。问题是我不知道：
目前有多少个NHibernate缓存了NHi
浏览 0提问于2010-06-09得票数 0
Akka程序更难调试。我们还需要跟踪许多参与者的状态来诊断问题。对于Flink为什么选择Akka有什么解释吗？
浏览 62提问于2019-09-09得票数 0
我正在从Kafka读取数据，并尝试以ORC格式将其写入HDFS文件系统。我使用了来自他们官方网站的以下链接参考。但我可以看到，Flink为所有数据写入完全相同的内容，并使如此多的文件和所有文件都可以103KB https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/connectors/streamfile_sink.html#orc-format 请在下面找到我的代码。 object BeaconBatchIngest extends StreamingBase {
val env: StreamExecutio
浏览 105提问于2020-07-10得票数 2
回答已采纳
1回答
由于性能测量，我想逐步执行为Flink编写的Scala程序。
execute first operator; materialize result;
execute second operator; materialize result;
...
诸若此类。原始代码：
var filename = new String("<filename>")
var text = env.readTextFile(filename)
var counts = text.flatMap { _.toLowerCase.split("\\W+") }.map { (
浏览 2提问于2015-11-13得票数 3
回答已采纳
1回答
我试图使用Kafka消息(作为StreamSource)更新Elasticsearch中的文档。使用windows和Elasticsearch连接器作为接收器在大容量中写入Elasticsearch是很好的，但是，我们需要更新文档中的现有数据并以大容量的方式读取它(不是针对每个元组，而是针对我们希望聚合的byKey()拆分后的整个窗口)。
我们现在使用的是Storm三叉戟，它在之前执行批量读取，然后将更新的聚合写回，最大限度地减少与后端的交互。我只是在Flink里找不到类似的东西-有什么线索吗？
浏览 1提问于2019-05-28得票数 1
回答已采纳
下面的代码片段将演示如何以流模式将数据写入文件。但是，当应用程序开始运行时，我检查了输出目录，发现没有写入数据，_SUCCESS也不存在
每个分区(例如：p_day=2021-04-30\p_hour=18\p_min=52)包含大约58个文件(每个文件是空文件，每秒钟创建一个文件)，名称类似于.part-01c10c8a-5ffa-4ffc-91d0-57ed68d85c93-0-0.inprogress.1cc628a9-0f88-46ca-9d14-7f10db332184。如果我用'format' = 'parquet',更改'format
浏览 0提问于2021-04-30得票数 1
我编写了Kafka流应用程序，我想将它部署到Kafka集群上。因此，我构建了一个jar文件并使用以下命令运行它：
java -jar KafkaProcessing-1.0-SNAPSHOT-jar-with-dependencies.jar testTopic kafka1:9092,kafka2:9092 zookeeper1:2181,zookeeper2:2181 output
它运行正常，但作业运行在我在命令上运行的机器上！我想当我指定BOOTSTRAP-SERVERS时，它会自动在集群上进行计算，而不是在主机上！
那么，我的问题是如何在Kafka集群上提交kafka流作业呢？就像
浏览 1提问于2017-12-01得票数 5
回答已采纳
谁能把Flink和Spark作为机器学习的平台进行比较呢？对于迭代算法，哪一种可能更好？链接到一般Flink与火花讨论：
浏览 3提问于2015-04-21得票数 27
回答已采纳
1回答
DSMS对应于数据流管理系统。这些系统允许用户提交将持续执行的查询，直到用户删除为止。
像Storm和Flink这样的系统能被看作是DSMS吗?或者它们是否更通用？
谢谢
浏览 8提问于2016-11-20得票数 2
回答已采纳
2回答
我想在配置单元中使用select查询一次获取一行。hive中不支持与SQL.But游标中的游标类似的内容。有没有办法实现这一点？
浏览 2提问于2013-07-09得票数 0
我想知道是否有可能编写一个在HBase上执行BulkLoad的java程序。我在hadoop集群上，但由于某些原因，我不需要编写MapReduce作业。
谢谢
浏览 1提问于2015-06-11得票数 0
1回答
我正在尝试构建涉及大量数据摄取的数据管理(DM)解决方案，通过一些数据域规则，替换(丰富)，在将错误数据发送到下游系统之前对其进行标记。规则检查和值替换可以是一些简单的东西，比如数据元素应该满足的允许阈值数值，也可以是一些更复杂的东西，比如对域值池的主数据进行查找。
您认为Apache Flink可以作为这种处理的一个很好的候选者吗？是否可以定义flink运算符来对流经它的每个元组进行查找(使用主数据)？对于后一个问题，我认为使用Apache Flink有一些缺点- 1)查找可能是一个阻塞操作，会降低吞吐量；2)如果操作员函数必须从其他地方获取主数据，则无法进行检查点和保持操作员状态。
他们的
浏览 10提问于2016-08-02得票数 0
支持多个运行程序后端，包括Apache和Flink。我很熟悉Spark/Flink，我正试图看到Beam批量处理的利弊。
从的角度来看，它觉得它非常类似于本机Spark/Flink的等价物，可能有一个稍微冗长一些的语法。
我目前不认为选择梁而不是火花/Flink来完成这样的任务有多大的好处。到目前为止，我能提出的唯一意见是：
支持:对不同执行后端的抽象。
Con:这种抽象的代价是减少了对在Spark/Flink中执行什么的控制。
是否有更好的例子突出梁模型的其他利弊？是否有任何关于失控如何影响性能的信息？
请注意，我并不是在请求流方面的差异，流方面的部分内容在中进行了介绍，并在
浏览 4提问于2017-04-24得票数 108
回答已采纳
我正在使用flink消费kafka并写入redis。
下面是我对redis的接收器函数：
.addSink(new RichSinkFunction<MobilePageEvent>() {
@Override
public void invoke(MobilePageEvent event, Context context) {
JEDIS_CLUSTER.zadd(..);
}
})
浏览 98提问于2019-02-28得票数 0
回答已采纳
我的powercenter版本是8.6.1。我设计了一个映射和一个会话，并在会话设置中选中了"insert“和"update else insert”。我发现数据处理的速度只有200行/秒左右。通过取消勾选"update else insert"，速度可以达到10000行/秒。目标表有一个索引，我检查了执行计划，索引有效。在DB (oracle)中执行update sql语句，不到一秒钟就更新了近一百万行。所以这不会是数据库的问题。有没有人遇到过同样的问题？我的会话设置有问题吗？任何帮助都是我的荣幸！
浏览 0提问于2016-12-07得票数 0
1回答
我只是尝试在Flink中使用Scala类型的类。我定义了以下类型的类接口：
trait LikeEvent[T] {
def timestamp(payload: T): Int
}
现在，我想考虑一下这样的DataSet of LikeEvent[_]：
// existing classes that need to be adapted/normalized (without touching them)
case class Log(ts: Int, severity: Int, message: String)
case class Metric(ts: Int, name:
浏览 5提问于2020-04-13得票数 2
回答已采纳
。 嗨, 使用Apache Flink 1.8。我有一个来自Kafka的JSON记录流，并对它们进行过滤，一切都很好。 现在，我想用一个数据库表中的查找值来丰富Kafka中的数据。 这是否只是创建两个流，加载第二个流中的表，然后连接数据的情况？ 数据库表确实会更新，但并不频繁，我希望避免在通过流的每条记录上查找DB。
浏览 24提问于2019-04-16得票数 5
回答已采纳
我已经在生产者端启用了快速压缩，批处理大小为64kb，并处理每个1KB的消息，并将逗留时间设置为inf，这是否意味着直到我处理64条消息，生产者才会将消息发送到kafka out topic...换句话说，producer会将每条消息发送给kafka，还是等待64条消息并分批发送……
因为偏移量是一个接一个地增加，而不是64的倍数。
编辑-使用flink-kafka连接器
浏览 0提问于2018-08-10得票数 1
3回答
我想在flink中缓冲一个数据流。我最初的想法是将100条数据缓存到一个列表或元组中，然后使用insert into values (???)将数据批量插入到clickhouse中。你有没有更好的方法呢？
浏览 36提问于2021-09-16得票数 1
1回答
我是第一次接触Tornado和Python Thread。我想要实现的是:我有一个龙卷风web服务器，接受用户的请求。我希望将一些数据存储在本地，并定期将其作为批量插入写入数据库。
import tornado.ioloop
import tornado.web
import threading
# Keep userData locally in memory
UserData = {}
def background(f):
"""
a threading decorator
use @background above the functi
浏览 1提问于2011-10-21得票数 4
我正试图遵循的官方教程。
安装先决条件之后，我执行以下命令：
$ mvn archetype:generate \
-DarchetypeGroupId=org.apache.flink \
-DarchetypeArtifactId=flink-walkthrough-datastream-java \
-DarchetypeVersion=1.15.0 \
-DgroupId=frauddetection \
-DartifactId=frauddetection \
-Dversion=0.1 \
-Dpackage=spendre
浏览 7提问于2022-06-24得票数 1
回答已采纳
1回答
Flink码头图像
、、、
我是新来的码头，我正在努力创建Flink码头形象。码头形象正在显示成功创建。但是我有一个疑问，我不能像在Dockerfile中提到的那样在/opt路径中看到Flink二进制文件。
如何知道我的Flink码头形象是成功创建的。
屏幕截图从控制台，很少一组命令是突出显示红色，意味着它可以吗？
创建显示Flink码头图像的状态：
请帮帮我。谢谢你..。
全日志：
sudo /home/develk/cntx_eng/build.sh \
> --job-artifacts /home/develk/cntx_eng/FlinkContextEnginePoc-0.0.1-S
浏览 2提问于2019-12-09得票数 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
我以独立模式在k8s上部署了几个flink，并通过一个promethus-pushgateway导出它们的指标。
问题是：
度量数据间歇性地到达promethus，当在grafana中显示时，会导致点间的间隙。
目标：
monitoring/pushgateway/0 (1/1 up)
Endpoint: http://172.19.88.111:9091/metrics
State : UP
Labels: endpoint="tcp" instance="172.19.88.111:9091" job="pushgateway" na
浏览 8提问于2021-05-24得票数 0
2回答
我正在尝试创建一个sbt Scala项目，添加一些来自Flink的代码。我已经成功地添加了除Flink-ML以外的大多数Flink项目。
我尝试添加不同的依赖项，错误如下：
sbt.build配置：
resolvers += Resolver.mavenLocal
// I have downloaded the Flink project and compiled in local.
libraryDependencies += "org.apache.flink" % "flink-ml_2.10" % "1.0-SNAPSHOT"
错误如下
浏览 2提问于2016-02-23得票数 1
Added Depedency Pom Details :
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.11</artifactId>
<version>1.7.1</version>
</dependency>
<depe
浏览 0提问于2019-04-04得票数 2
回答已采纳
根据BEAM的，并根据许多，join可以通过CoGropByKey或KeyedPCollectionTuple ()来实现。
没有人在谈论这种转变的表现。
我的流程应该非常简单:从be表(TableRow)输入行批，并通过相同的键将它们(或“充实”)与其他be表中的其他值连接起来。因此，最终的输出也应该是TableRow类型的。
我想了解什么将是最好的做法，将2张桌子在梁与烧烤桌。
例如，我可以在BQ中创建一个视图，替换所有这些管道，并执行更高效的联接操作，但我希望处理代码中的所有逻辑
当join操作正在处理时，引擎盖下发生了什么？
为了加入所有管道批处理(一项接一项)，DirectRunne
浏览 0提问于2019-06-19得票数 1
我需要将表结果保存到S3上的orc中，我就是这样做的：
tEnv.createTemporaryTable("my_output_table", TableDescriptor.forConnector("filesystem")
.schema(outputSchema)
.option("path", s3OutputPath)
.format(FormatDescriptor.forFormat("orc").build())
浏览 15提问于2022-10-17得票数 0
回答已采纳
我想使用Flink对Postgresql中的数据进行一些分析。我应该如何和在哪里给出端口地址，用户名和密码。我正在尝试使用链接：中提到的表源。
final static ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
final static TableSource csvSource = new CsvTableSource("localhost", port);
实际上，我无法开始。我翻阅了所有的文件，但没有找到详细的报告。
浏览 0提问于2018-06-21得票数 1
回答已采纳
我想在我的工作中添加一个JVM选项。因此，我在flink- line . the中添加了以下一行：
env.java.opts: "-Dspring.config.location=/root/application.properties"
当我使用命令./start-local.sh启动flink时，jobmanager和taskmanager很好地加载了JVM选项：
JobManager日志：
org.apache.flink.runtime.jobmanager.JobManager - JVM Options:
org.a
浏览 0提问于2018-03-01得票数 0
回答已采纳
我正在尝试创建到oracle数据库的Flink JBDC接收器。当在本地运行时(从junit测试和minicluster)，它可以工作，但是当部署在k8s中时，它会抛出一个异常，说它找不到合适的驱动程序。Classpath是：
Classpath: /flink/lib/flink-cep-scala_2.12-1.13.5-stream1.jar:/flink/lib/flink-connector-jdbc_2.12-1.13.5.jar:/flink/lib/flink-csv-1.13.5-stream1.jar:/flink/lib/flink-json-1.13.5-stream1
浏览 16提问于2022-05-24得票数 1
回答已采纳
我有一个容器化的flink集群，具有独立的JobManager和2个任务管理器。当我提交一个包含我需要的连接器和代码的胖jar时，一切都会正常运行，但是一个只有我的代码的jar将失败--尽管我已经将连接器jar添加到/opt/flink/lib，容器进程显示它们将被添加到类路径中。当我试图从IDE进行远程连接时，也会发生同样的失败；不用说，每次都要打包一个胖jar，这会带来糟糕的开发体验。
我需要做什么才能让flink理解我放置在/opt/flink/lib中的额外jars应该与所有新作业一起使用？
类加载程序异常：
org.apache.flink.streaming.runtime.ta
浏览 0提问于2019-04-12得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
腾讯会议活动推荐
腾讯云开发者
