Flink中的`stream.addSink(new PrintSinkFunction<>())`和`stream.print()`是否相同
在Flink中，stream.addSink(new PrintSinkFunction<>())和stream.print()都用于将流数据打印到标准输出。它们实现的功能是相同的，都是为了方便开发者在调试阶段查看流数据的内容。
stream.addSink(new PrintSinkFunction<>())是将流数据通过PrintSinkFunction输出到标准输出，可以将流数据打印到控制台或日志文件中，以便开发者查看。在实际应用中，可以根据需要自定义PrintSinkFunction来实现不同的输出方式。
stream.print()也是将流数据打印到标准输出，但它是Flink提供的简化版打印方法。使用stream.print()时，Flink会默认使用PrintSinkFunction来输出流数据，无需手动指定。它可以直接将流数据打印到控制台，方便开发者快速查看流数据的内容。
综上所述，stream.addSink(new PrintSinkFunction<>())和stream.print()在功能上是相同的，都用于将流数据打印到标准输出，只是使用方式稍有差异。在实际开发中，可以根据个人习惯或具体需求选择使用其中之一。
推荐的腾讯云相关产品：
- 腾讯云流计算 Flink：提供了基于Flink的流式计算服务，支持实时数据处理、流式ETL等场景。详情请参考：腾讯云流计算 Flink
- 腾讯云云原生应用平台 TKE：提供了Kubernetes托管服务，适用于云原生应用的部署和管理。详情请参考：腾讯云云原生应用平台 TKE
相关·内容
我看过stream.addSink(new PrintSinkFunction<>())和stream.print()，不知道它们之间有什么不同。 基于https://github.com/apache/flink/blob/b2a342c6a6ef154ed3c1a44826ce2be14e538386/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java#L972-L973，除了.name("Print to Std. Out")
浏览 118提问于2020-10-30得票数 0
回答已采纳
我试图在每个JSON上创建一个500 ms数据集，并希望将其推送到Kafka主题，这样我就可以在下游设置一些窗口并执行计算。下面是我的代码：
package KafkaAsSource
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
i
浏览 3提问于2021-12-26得票数 2
回答已采纳
3回答
关于Flink反序列化的两个问题
、、、、
我是Flink和集群计算的新手。我花了一整天的时间试图正确地解析Flink，一条来自Kafka的愚蠢的流，没有结果:这有点令人沮丧.在kafka中，我用一个字符串键标识了一个JSON-LD消息流。我只想在Flink中检索它们，然后用不同的键分离消息。
1)最初，我考虑以字符串的形式发送消息，而不是JSON。我觉得更容易..。
我试过所有的去序列化器，但都没有用。简单的反序列化器不明显地起作用，但它完全忽略键。
我认为我必须使用(Flink显然只有两个支持键的反序列化器)：
DataStream<Object> stream = env
.addSource(
浏览 8提问于2017-06-12得票数 1
我想对Spark和Flink进行基准测试，为此我做了几个测试。然而，Flink不适用于Kafka，而Spark适用于完美。
代码非常简单：
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
val properties = new Properties()
properties.setProperty("bootstrap.servers", "localhost:9092")
properties.setProperty("
浏览 0提问于2016-06-16得票数 0
我正在尝试将Flink连接到Kafka消费者
我正在使用Docker Compose构建4个容器zookeeper、kafka、Flink JobManager和Flink TaskManager。
对于zookeeper和Kafka，我使用的是wurstmeister图像，而对于Flink，我使用的是官方图像。
docker-compose.yml
version: '3.1'
services:
zookeeper:
image: wurstmeister/zookeeper:3.4.6
hostname: zookeeper
expose:
浏览 0提问于2019-02-24得票数 14
首先，我在Kafka topic中加载了一个CSV文件，我可以通过中的Flink程序打印该主题。代码如下： final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
Properties prop = new Properties();
prop.setProperty("bootstrap.servers",
"10.32.0.2:9092,10.32.0.3:9092,10.32.0.4:9092"
浏览 16提问于2019-06-24得票数 0
回答已采纳
我使用了Apache Bahir的监听ActiveMQ的AMQSource连接器，但是当我运行Flink作业来使用来自ActiveMQ的数据时，没有生成任何输出。 例如，连接器正在侦听包含4条消息的ActiveMQ，但当我运行Flink作业时，不会消耗任何数据。 val brokerURL = "tcp://localhost:61616"
val destinationName = "TEST.FOO"
val filePath = "C:\\output" + System.currentTimeMillis + ".csv"
浏览 51提问于2019-02-28得票数 0
我遵循()并编写了下面的代码片段，用一个简单的TextInputFormat在dir中处理.gz日志文件。它在我的本地测试目录上工作，扫描并自动打开.gz文件内容。但是，当我使用s3桶源运行它时，它不会处理.gz压缩文件。不过，这个Flink作业仍然会在.log桶上打开s3文件。似乎它只是不解压缩.gz文件。如何在s3文件系统上解决这个问题？
public static void main(String[] args) throws Exception {
final ParameterTool params = ParameterTool.fromArgs(args);
f
浏览 6提问于2021-12-21得票数 1
回答已采纳
我通过下面的说明创建了一个主题：
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
回答已采纳
1回答
package org.apache.spark.examples.kafkaToflink;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintStream;
import java.nio.charset.StandardCharsets;
import java.util.Properties;
import org.apache.flink.streaming.api.datastream.DataStre
浏览 3提问于2017-04-19得票数 1
回答已采纳
我试图使用来自s3的Flink读取IntelliJ中的一个文件，并获得以下异常：
原因: java.lang.ClassNotFoundException:未找到类org.apache.hadoop.fs.s3a.S3AFileSystem
我的代码是这样的：
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.s
浏览 0提问于2021-03-02得票数 1
1回答
我开发了一个简单的Flink作业，它消耗了卡夫卡流。在当地环境方面，大家都做得很好。但是，当我试图用纱线在集群上运行我的Flink工作时，什么都不会发生。(而且，我没有任何错误消息)。
./bin/flink run -m yarn-cluster -yn 4 /Flink/Flink_Stream_Kafka/jar-with-dependencies.jar --server X.X.X.X:9092 --topic logs
当然，我已经检查了我的集群是否能够访问流，并且它已经访问了。我甚至可以在集群的每台机器上作为一个简单的java程序运行我的工作，它可以工作。
有什么能解释这事的见解
浏览 5提问于2017-03-23得票数 0
回答已采纳
我想启动一个RabbitMQ源，然后接收，但我不能执行第一步，即启动Rabbit MQ源。RabbitMQ服务器正在运行，我也可以看到仪表板。
我的代码如下
公共类rabbitmq_source {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment envrionment = StreamExecutionEnvironment.getExecutionEnvironment();
RMQConnectionConfig connectionConf
浏览 2提问于2018-01-21得票数 3
我正在为工作中的一个项目尝试flink。我已经达到了通过应用计数窗口等来处理流的地步。然而，我注意到了一种奇怪的行为，我无法解释。
看起来一个流是由两个线程处理的，输出也分成两个部分。
首先，我注意到使用stream.print()将流打印到标准控制台时的行为。
然后，我打印到一个文件，它实际上打印在输出文件夹中名为1和2的两个文件中。
SingleOutputStreamOperator<Tuple3<String, String,String>> c = stream_with_no_err.countWindow(4).apply(new CountPer
浏览 3提问于2017-02-24得票数 2
回答已采纳
我试着测试简单的flink卡夫卡例子。mvn package工作正常，然后运行../../flink-1.14.3/bin/flink run -c com.comapny.flinktest.App target/flinktest-1.0-SNAPSHOT.jar并得到以下错误：
java.lang.NoClassDefFoundError: org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer
at com.optiver.flinktest.App.main(App.java:38)
at sun.
浏览 1提问于2022-02-22得票数 0
1回答
我正在尝试使用Apache阅读和打印来自Kafka的Protobuf消息。
我跟踪了官方文档，但没有成功：
Flink消费者代码是：
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
env.enableCheckpointing(5000)
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
浏览 0提问于2022-01-03得票数 0
回答已采纳
我想在卡夫卡上设置一个基本的生产者-- Flink的消费者，但是我在通过Java向现有消费者生成数据时遇到了问题。
CLI解决方案
我通过命令使用来自Kafka broker的kafka_2.11-2.4.0 zip设置了一个https://kafka.apache.org/downloads
bin/zookeeper-server-start.sh config/zookeeper.properties
和bin/kafka-server-start.sh config/server.properties
我创建了一个名为transactions1的主题
bin/kafka
浏览 4提问于2020-03-09得票数 0
回答已采纳
我刚刚开始使用Apache (Scala )，我的问题是:基于Flink站点的一个例子，我尝试将Kafka的数据流到Apache中：
val stream =
env.addSource(new FlinkKafkaConsumer09("testing", new SimpleStringSchema() , properties))
一切正常工作，stream.print()语句在屏幕上显示以下内容：
2018-05-16 10:22上午1:44
为了加载数据，我想使用case类，我已经尝试使用
flatMap(p=>p.split("|"))
浏览 0提问于2016-05-17得票数 0
回答已采纳
1回答
flink中的弹簧自动配置
、、、、
我试图使用的组合flink和跳板，我有一些问题。让我们说，我有这样的流动。
获取包含日期字符串的json字符串。使用map函数和ObjectMapper将其解析为的对象。
这是一个简单的用法，可以描述我的行为。
所以，我有包含LocalDateTime字段的Word类表示词。
@Data
public class Word {
@JsonDeserialize(using = LocalDateTimeSerde.class)
LocalDateTime date;
}
LocalDateTimeDeserlization看起来就是这样(我想自动配置应用程序)：
@Requir
浏览 0提问于2020-01-29得票数 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
1回答
Apache多个输出行
、、、
我试图按下面的方式运行flink作业来读取Apache &print中的数据：
Java程序
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "test.net:9092");
properties.setProperty(&#
浏览 1提问于2017-03-28得票数 0
回答已采纳
我有一个运行在Flink runner上的光束管道，如下所示： ? 光束中的整个管道都被捆绑到这个区块中。为什么Flink用这种方式来说明它？ ? 当然，管道步骤中的不同pardo应该由不同的块和工作者来表示(就像在Google的数据流中一样)。是否需要在Flink运行器中设置某些内容以显示更细粒度的任务，或者Flink UI是否仅为原生Flink代码构建，而这正是该UI处理Apache光束管道的方式？
浏览 24提问于2021-07-05得票数 1
2回答
我试图从我的Java程序中编写一个文件，但是什么也没有发生。我没有收到任何异常或错误，它只是默默地失败了。
try {
File outputFile = new File(args[args.length - 1]);
outputFile.delete();
outputFile.createNewFile();
PrintStream output = new PrintStream(new FileOutputStream(outputFile));
浏览 0提问于2009-11-20得票数 0
回答已采纳
1回答
我试图写一个卡夫卡消费者，它消耗来自一个主题的数据。但是，每当我尝试运行它时，我都会得到以下错误。
Exception in thread "main" org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$6.apply$mcV$sp(JobManager.scala:897)
at or
浏览 0提问于2018-05-18得票数 2
回答已采纳
我尝试使用flink的表和sql作为一个简单的示例，在这个示例中，我从文件中读取字符串，将其转换为Tuple2，并尝试将其插入表中。这是我的密码。
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.table.StreamTableEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.table.Table;
import org.ap
浏览 1提问于2016-12-21得票数 2
1回答
我现在的情况是，我想停止/取消代码中的flink作业。这是在我的集成测试中，我将一个任务提交给我的flink工作并检查结果。当作业异步运行时，即使测试失败/通过，它也不会停止。我想在考试结束后把车停下来。
我尝试了以下几种方法：
找工作经理演员
得到运行的作业
对于每个正在运行的作业，向作业管理器发送一个取消请求。
当然，在没有运行的情况下，我不确定这个工作经理是错误的，还是缺少了其他的东西。
我得到的错误是:flink-akka.actor.default-5消息org.apache.flink.runtime.messages.JobManagerMessages$Re
浏览 1提问于2017-02-09得票数 4
我使用Flink 1.7.2编写了一个简单的字数统计应用程序，同时使用Kafka 2.2作为消费者和生产者。我对Kafka生产者使用只有一次的语义，使用KeyedProcessFunction进行有状态处理，使用MapState保持我的状态，使用增量检查点的RocksDB作为我的状态后端。
当我在IntelliJ上运行这个应用程序时，它工作得很好，但是当我把它提交到我本地的Flink集群时，我收到了AsynchronousException异常，Flink应用程序每隔0-20秒就会重试一次。以前有没有人遇到过这个问题？从配置的角度看，我是否遗漏了什么？
下面是我的代码：
class Keyed
浏览 37提问于2019-07-05得票数 0
回答已采纳
我想在我的Flink作业(Scala)中跨Flink的并行任务共享一个私有变量。我的代码是这样的：
object myJob extends flinkJob {
private val myVariable = someValue
def run(params) : Unit = {
//Stream processing
//myVariable is used here in the ProcessFunction
}
}
当我使用一些并行化运行这个作业时，是否会在所有的myVariable任务之间共享一个“Flink”副本？如果不是，如何确保在所有并行任务中只使用和维护
浏览 4提问于2019-04-04得票数 1
我有一个Cassandra数据库，它必须在我的Flink程序中从套接字中接收它的数据，比如用于流处理的蒸汽。因此，我编写了一个简单的客户端程序，它从Cassandra读取数据并将数据发送到套接字；此外，我在服务器base.In事实中编写了Flink程序，我的客户端程序很简单，不使用任何Flink指令；它只向套接字发送字符串格式的Cassandra行，服务器必须接收该行。首先，运行Flink程序来侦听客户端，然后运行客户端程序。客户端从服务器接收到此流(因为服务器发送数据，而客户端无法正确接收)：
嗨客户端org.apache.flink.streaming.api.datastream.Dat
浏览 0提问于2018-12-27得票数 0
回答已采纳
我们目前正在Kubernetes上以独立模式运行flink集群。我们希望探索是否可以迁移到AWS (KDA)上的托管flink。
但是，我似乎没有发现任何文档或迹象表明可以注入环境变量？这些是否需要作为运行时参数提供？
与之相关的是，是否有可能覆盖我们当前在托管Flink中的flink-conf.yml . our中指定的默认flink配置？
提前感谢！
浏览 16提问于2022-08-25得票数 1
下面是使用Flink和Kafka的例子。我只找到像this page这样的结果，它们不能正确编译，并且给出了很难查找的错误消息。 基本上，当我试图编译这段代码时，我得到了错误： import org.apache.flink.streaming.util.serialization.DeserializationSchema;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafka
浏览 170提问于2019-05-24得票数 2
回答已采纳
尝试使用Flink读取"avro“序列化数据的Kafka流，如下所示： tableEnv.connect(new Kafka()
.version("0.11")
.topic(source.getTopic())
.properties(source.getProperties())
.startFromLatest())
.withSchema(Schemafy.getSchemaFromJs
浏览 359提问于2019-06-05得票数 2
我想流式传输一个csv文件，并使用flink执行sql操作。但是我写的代码只读一次就会停止。它不会流式传输。提前谢谢你，
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.getTableEnvironment(env);
CsvTableSource csvtable = CsvTableSource.builder()
.path("D
浏览 11提问于2017-07-25得票数 4
回答已采纳
2回答
首先，这与非常相似，但并不相同。这个问题的答案似乎解决不了我的问题。如果我在答案中遗漏了什么，那么请重新表述答案，因为我显然遗漏了一些东西。
问题是完全一样的，尽管-- Flink (卡夫卡连接器)重新运行它在关闭之前看到的最后3-9条消息。
我的版本
Flink 1.1.2
Kafka 0.9.0.1
Scala 2.11.7
Java 1.8.0_91
我的密码
import java.util.Properties
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.strea
浏览 4提问于2016-09-12得票数 7
回答已采纳
我在这里设置了一个最小的例子，其中我有N个来自N Kakfa主题的流(在下面的例子中是100)。
当看到"EndofStream“消息时，我想要完成每个流。当所有流都完成时，我期望Flink程序能够优雅地完成。
当并行设置为1时，这是正确的，但一般情况下不会发生。
从来看，似乎不是卡夫卡消费者群体的所有线索都结束了。
建议抛出一个异常。但是，程序将在第一个异常时终止，不会等待所有流完成。
我还添加了一个最小的python程序来向kafka主题添加信息，以提高可重复性。请在每个程序中填写<IP>:<PORT>。
final StreamExecut
浏览 0提问于2020-07-01得票数 2
其中Maven包是
我经常在示例中找到带有导入语句的示例，但是在这个特定的例子中，搜索Maven或搜索引擎也没有给出结果。
下面是我得到的错误的图片：
pom.xml中的Flink包是：
flink-java
flink-streaming-java_${scala.binary.version}
flink-clients_${scala.binary.version}
flink-statebackend-rocksdb_2.11
flink-table-api-java
flink-sql-client_2.11
flink-test-utils_2.11
我还试图包括：
flin
浏览 23提问于2022-08-02得票数 0
回答已采纳
问题
在不同的包中ProcessWindowFunction的区别是什么?在哪里/什么时候使用哪一个包？
org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction
org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
对于窗口操作，就像在org.apache.flink.streaming.api.scala中一样，是正确使用的，但是找不到FLink1.8API文档，只有1.3。
如果我使用org
浏览 1提问于2019-07-07得票数 0
1回答
让我把它说出来，我是一个非常初学者的Flink，并试图抓住尽可能多的概念。
比方说，我有一个flink集群，其中包含10个任务管理器。我每个人都有一个flink作业在运行。作业也使用广播状态。该广播状态是通过每10分钟读取5个S3文件，进行一些处理，并创建播放的int to list of strings地图来创建的。
问题:在哪里读取文件，是否在JobManager读取和处理文件，并将处理过的内容发送给任务管理人员。
或
是任务管理人员负责所有的读取和处理。如果是这种情况，那么flink如何确保如果任务管理器无法从S3读取，那么所有任务管理器的广播状态都是相同的。
编辑
因此，任务管理器读取
浏览 2提问于2019-10-21得票数 0
对于KeyedStream#timeWindow#process，我想知道一个窗口实例是否只包含相同的键，不同的键是否使用不同的窗口实例。
从以下应用程序的输出中，我看到一个窗口实例将只包含相同的键，不同的键将使用不同的窗口。
但我想问一下并确认，谢谢！
import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
import scala.util.Random
class KeyByAndWindowAndProcessTestSource exte
浏览 1提问于2019-09-14得票数 0
回答已采纳
我正在编写一个如下所示的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中的一个输入数据流应用ProcessFunction()，用一个缓存对象处理每个传入的元素。我的代码看起来像这样： object myJob extends FlinkJob {
private val myCache = InMemoryCache()
private def updateCache(myCache,someValue) : Boolean = {//some code}
private def getValue(myCache,someKey) : Boolean = {//some code}
def run(params, executi
浏览 41提问于2019-04-05得票数 0
回答已采纳
我想把Flink和卡桑德拉联系起来。我正在从运行示例程序，但是我得到了以下错误：
线程"main“中的异常: org.apache.flink.streaming.api.datastream.DataStreamSink.name(Ljava/lang/String;)Lorg/apache/flink/streaming/api/datastream/DataStreamSink；at org.apache.flink.streaming.connectors.cassandra.CassandraSink$CassandraPojoSinkBuilder.build(Cassa
浏览 19提问于2016-10-20得票数 1
我使用的是Flink 1.15.1和JUnit5。我想知道从中改编的下面的集成测试是如何工作的，而不包含flink-test-utils依赖项和MiniClusterWithClientResource静态实例。
package com.mypackage;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.apache.flink.stre
浏览 38提问于2022-08-09得票数 0
回答已采纳
1回答
我试图在Flink (1.4.2版)上使用可查询状态，但不幸的是，我一直收到以下错误：
INFO my.test.flink.QueryableState - Params are a96438fa12879b7598c9cf32684e2669, kafka-cluster_jobmanager_1, 6123
INFO my.test.flink.QueryableState - Before the call java.util.concurrent.CompletableFuture@26aa12dd[Not completed]
java.util.concurrent.Ex
浏览 5提问于2018-05-18得票数 0
我在Arduino IDE上使用Raspberry pi pico。我正在使用这个库作为它。在这个链接中有3个例子，和没有打印任何东西。艾德说
WARNING: library ArduinoUniqueID claims to run on avr, esp8266, esp32, sam, samd, stm32 architecture(s) and may be incompatible with your current board which runs on mbed_rp2040 architecture(s).
(但GitHub说我们添加了RP2040)
当我尝试使用最后一个示
浏览 9提问于2022-06-12得票数 2
回答已采纳
我正在尝试运行流作业，它消耗来自Kafka的消息，转换它们，并将其下沉到Cassandra。
当前代码片段失败
val env: StreamExecutionEnvironment = getExecutionEnv("dev")
env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
.
.
.
.
val source = env.addSource(kafkaConsumer)
.uid("kafkaSource")
浏览 2提问于2020-05-27得票数 1
最近，我遇到了一些关于开发flink作业的问题，它引入了spring和hibernate，作业将在flink集群上运行。因此，在任务管理器上运行flink运算符之前，需要初始化spring资源，而不是作业manager.But，我找不到合适的StreamExecutionEnvironment方法来实现。
以下是我尝试过的一些方法：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
ApplicationContext context = new ClassPathXml
浏览 4提问于2016-11-11得票数 4
1回答
我试图了解如何使用RabbiMQ的Flink中的高可用性，使用相关is和启用检查点，但这是行不通的。我的制片人代码：
connection = factory.newConnection();
Channel channel = connection.createChannel();
String corrId = java.util.UUID.randomUUID().toString();
BasicProperties props = new AMQP.BasicProperties().builder().correlationId(corrId).build();
channel
浏览 2提问于2016-08-30得票数 3
回答已采纳
我是Flink和EMR集群部署的新手。目前，我们有一个Flink作业，我们正在通过Flink CLI停止/启动作业命令在AWS EMR集群上手动部署它。
我想要自动化这个过程(在通过保存点的管道进行的每个部署中自动更新flink作业jar )，并需要一些关于可以探索的方法的建议。
浏览 3提问于2022-04-22得票数 0
我正在设法将一些文件添加到Flink中的执行器中。就像星火中的SparkContext.addFile() SparkFiles.get()一样，Flink中是否有任何可能的API？谢谢。
浏览 7提问于2020-01-04得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
