flink流批开源
您好！感谢您的提问。FLINK是一个流行的开源大数据处理框架，全称为Apache Flink。它是一个高性能、可扩展、分布式的数据处理引擎，可以处理实时流数据和批量数据。FLINK支持多种编程语言，包括Java、Scala和Python等。
FLINK的优势在于它的低延迟和高吞吐量，可以快速处理大量的数据。它还具有容错能力，可以在故障发生时恢复数据，并且可以与其他大数据处理框架集成，例如Apache Kafka和Apache Hadoop等。
FLINK的应用场景非常广泛，包括实时数据分析、数据流处理、数据转换、数据聚合、数据清洗、数据存储等。
腾讯云提供了FLINK的相关产品，可以满足用户的大数据处理需求。腾讯云提供的产品包括TDSQL、TDID、TDRDS、TDTS等，可以满足不同用户的需求。腾讯云的产品都是基于云计算技术，可以快速部署和使用，同时也提供了丰富的文档和教程，帮助用户快速上手。
希望我的回答能够帮助您解决问题。如果您还有其他问题，欢迎随时提问。
相关·内容
2回答
我们知道Flink是一个真正的实时流处理引擎，它可以在记录到达时处理记录，我们也知道火花流是一个微批量流处理引擎。
然而，我们也知道星星之火发布了结构化的流媒体，它怎么样？它是否也是一个真正的实时流处理引擎，就像Flink一样，它可以在记录到达时立即处理记录，而不是微批处理，或者仍然使用微批处理模式？
浏览 3提问于2017-06-27得票数 3
回答已采纳
1回答
如何将PyFlink性能与Flink + Scala进行比较？
大照片。我们的目标是建立冷和热层的Lambda架构。冷(批)层将用Apache (PySpark)实现。但是对于热(流)层有不同的选择:火花流或Flink。
因此，Apache是纯流的，而不是Spark的微批，我倾向于选择Apache。但我唯一关心的是PyFlink的性能。PySpark流的延迟会更少吗？它比Scala编写的Flink代码慢吗？在什么情况下更慢？
提前谢谢你！
浏览 3提问于2021-11-05得票数 1
回答已采纳
我使用的是Flink v1.4.0。我使用如下代码所示的Kafka FLink Consumer来使用Kafka主题中的数据：
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
// only required for Kafka 0.8
properties.setProperty("zookeeper.connect", "localhost:2181");
prope
浏览 6提问于2018-01-25得票数 3
回答已采纳
1回答
我正在尝试使用Flink以流的热情从消息队列中使用一个有限制的数据。数据将采用以下格式：
{"id":-1,"name":"Start"}
{"id":1,"name":"Foo 1"}
{"id":2,"name":"Foo 2"}
{"id":3,"name":"Foo 3"}
{"id":4,"name":"Foo 4"}
{"id":
浏览 0提问于2021-11-03得票数 0
回答已采纳
以下用例的最佳实践建议是什么？我们需要将流与一组“规则”相匹配，这些“规则”本质上是一个Flink DataSet概念。对此“规则集”的更新是可能的，但并不频繁。每个流事件必须对照“规则集”中的所有记录进行检查，而且每次匹配都会将一个或多个事件生成到接收器数据流中。规则集中的记录数在6位范围内。
目前，我们只是将规则加载到本地规则列表中，并在传入的flatMap上使用DataStream。在flatMap中，我们只是迭代一个列表，将每个事件与每个规则进行比较。
为了加快迭代速度，我们还可以将列表分成几个批，实质上是创建一个列表列表，并创建一个单独的线程来迭代每个子列表(使用Java或Scala
浏览 1提问于2018-06-04得票数 0
回答已采纳
我试图用Flink 1.5.0创建一个面向批处理的Flink作业，并希望使用Table和SQL来处理数据。我的问题是试图创建一个编译错误的BatchTableEnviroment
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment无法访问BatchJob.java:46,73
引起于
final BatchTableEnvironment bTableEnv = TableEnvironment.getTableEnvironment(bEnv);
据我所知，我不依赖于流环境。我的代码是下面的代码片
浏览 1提问于2018-06-19得票数 0
2回答
在使用Scala代码将数据从输出NiFi端口传输到Flink方面，我需要一些帮助。
我被.addSource()函数卡住了。它要求额外的参数(OUT)，但是当我提供它们时，我总是得到一个错误。Scala代码和错误消息如下所示。
package flinkTest
import java.nio.charset.{Charset, StandardCharsets}
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api
浏览 0提问于2017-07-17得票数 0
回答已采纳
我正在寻找一个可视化的编程工具，以抽象化的代码，当创建数据流流，以建立在Flink或束-Flink中的CEP应用程序。类似于Nifi或流设置数据收集器的图形设计器。
有人知道这样的计划吗？
浏览 5提问于2017-04-30得票数 4
回答已采纳
1回答
为了丰富数据流，我们计划将MySQL (MemSQL)服务器连接到现有的flink流应用程序
正如我们所看到的，Flink提供了一个带有JDBC连接器https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/jdbc/的Table
此外，我还发现了另一个名为Flink-CDC https://ververica.github.io/flink-cdc-connectors/master/content/about.html的MySQL连接器，允许以流的方式处理外部数据库。
他们之间有什么区别？在我的情况
浏览 19提问于2022-02-07得票数 0
回答已采纳
我有一个像这样的流：<_time(timestamp), uri(string), userId(int)>。_time属性是rowtime，我将它注册为一个表：
tableEnv.registerDataStream("userVisitPage", stream, "_time.rowtime, uri,userId");
然后我查询表：
final String sql =
"SELECT tumble_start(_time, interval '10' second) as timestart,
浏览 1提问于2018-03-02得票数 1
1回答
我正在研究Flink中join的各种实现。在批处理模式下，我遇到了hybrid-hash join和sort-merge join。在这两种情况下，都有一个阻塞洗牌，它是在连接之前完成的，因此连接之前的操作符的输出被实现到一些非临时存储中，如所说的。
我现在正在查看流连接的情况。我看到了一个实现，其中为两个输入创建了两个哈希表。每当输入出现时，都会将其保存在哈希表中，并对其他哈希表进行探测以产生结果。为了限制哈希表的大小，我们在哈希表中放置了一个输入保存的窗口。我的第一个问题是：
Do all stream join cases have this requirement of a windo
浏览 1提问于2021-05-14得票数 0
回答已采纳
支持多个运行程序后端，包括Apache和Flink。我很熟悉Spark/Flink，我正试图看到Beam批量处理的利弊。
从的角度来看，它觉得它非常类似于本机Spark/Flink的等价物，可能有一个稍微冗长一些的语法。
我目前不认为选择梁而不是火花/Flink来完成这样的任务有多大的好处。到目前为止，我能提出的唯一意见是：
支持:对不同执行后端的抽象。
Con:这种抽象的代价是减少了对在Spark/Flink中执行什么的控制。
是否有更好的例子突出梁模型的其他利弊？是否有任何关于失控如何影响性能的信息？
请注意，我并不是在请求流方面的差异，流方面的部分内容在中进行了介绍，并在
浏览 4提问于2017-04-24得票数 108
回答已采纳
1回答
我正在尝试将Flink stateful函数部署为flink jar，并遵循了here的说明。然而，我得到的错误是，即使我在pom中添加了依赖项，程序入口点类也没有在jar中找到。 运行： bin/flink run -c org.apache.flink.statefun.flink.core.StatefulFunctionsJob flink-statefun/statefun-examples/statefun-greeter-example/target/statefun-greeter-example-2.3-SNAPSHOT.jar 产生以下错误： org.apache.fli
浏览 237提问于2020-12-23得票数 2
这段代码将pandas转换为flink表，进行转换后再转换回pandas。当我使用filter filter而不是select时，它工作得很好，但当我添加group_by和order_by时，它会给我一个错误。 import pandas as pd
import numpy as np
f_s_env = StreamExecutionEnvironment.get_execution_environment()
f_s_settings = EnvironmentSettings.new_instance().use_old_planner().in_streaming_mode().
浏览 102提问于2020-07-16得票数 0
回答已采纳
1回答
我正在研究使用Flink作为流媒体引擎。我来自apache-storm，据我所知，storm的Bolt类似于Flink的任务/操作符。在风暴中，一个人可以 builder.setBolt("TEST", new TestBolt(),5)
.fieldsGrouping("Source1", "ID1")
.fieldsGrouping("Source2","ID2)
.fieldsGrouping("Source3","ID1")
浏览 32提问于2021-07-20得票数 0
我试图使用flink 1.15.1编写一个简单的表API S3流接收器(csv格式)，并面临以下例外：
Caused by: org.apache.flink.util.SerializedThrowable: S3RecoverableFsDataOutputStream cannot sync state to S3. Use persist() to create a persistent recoverable intermediate point.
at org.apache.flink.fs.s3.common.utils.RefCountedBufferingFileStream
浏览 20提问于2022-07-11得票数 0
回答已采纳
我想加入两个流(json)来自卡夫卡制片人。如果我过滤数据，代码就能工作。但当我加入他们的时候，它似乎不起作用。我想要打印到控制台的连接流，但什么也没有出现。这是我的密码
import java.util.Properties
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010
import org.apache.flink.streaming.util.serialization.SimpleStringS
浏览 2提问于2017-08-03得票数 1
回答已采纳
我有以下针对flink 1.12.0的简单代码，当我运行它时，异常抛出，有人能帮我看一下吗？谢谢!
package org.example.model
import java.sql.Timestamp
case class Person2(name:String, birthday: Timestamp)
和
package org.example.sql2
import java.sql.Timestamp
import java.util.Date
import org.apache.flink.streaming.api.TimeCharacteristic
import or
浏览 2提问于2020-12-25得票数 0
回答已采纳
我试图在flink服务器上运行一些代码，但无法
下面是我的代码
DataStream<UserInfo> keyedStream = executionEnvironment
.addSource(new UserDataSource());
keyedStream.assignTimestampsAndWatermarks(new MessageWaterEmitter());
tableEnv.registerDataStream("test", keyedStream, "us
浏览 0提问于2020-12-30得票数 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
我在使用Flink连接两个流的上下文中查看，并希望了解这两个流之间的区别以及影响Flink处理它们的方式。
作为一个相关的问题，我还想了解CoProcessFunction与KeyedCoProcessFunction有何不同。
浏览 2提问于2021-02-17得票数 1
回答已采纳
我使用intellij、maven 3和flink 1.15.1编写有状态流作业。我正在尝试为我的自定义KeyedProcessFunction编写单元测试，并尝试遵循文档以及添加提到的的依赖项。我对使用KeyedOneInputStreamOperatorTestHarness类很感兴趣，但是在我的任何依赖项中都找不到这个类(下面发布的修改后的pom )。我唯一能找到这门课的方法就是
<dependency>
<groupId>org.apache.flink</groupId>
<ar
浏览 8提问于2022-08-01得票数 1
回答已采纳
为什么Flink-HBase connector中没有HBaseReadStreamExample 似乎有一个读、写和WriteStream示例，但没有ReadStream。 从HBase到Flink的流读取是一种糟糕的做法吗？
浏览 21提问于2020-07-03得票数 1
回答已采纳
我有两个流。第一个是基于时间的流，我使用countTimeWindow接收前10个数据点来计算STAT值。我手动使用了变量cnt来只保留第一个窗口，并过滤了剩余的值，如下面的代码所示。 然后，我想使用这个值来过滤主流，以便获得比我在窗口流中计算的stat值更大的值。 然而，我不知道如何合并或计算这两个流来实现我的目标。 我的场景是，如果我将第一个stat值转换为广播变量，然后将其传递给主流，以便能够根据广播变量中的stat值过滤传入的值。 下面是我的代码。 import com.sun.org.apache.xpath.internal.operations.Bool;
import org
浏览 10提问于2019-02-12得票数 0
回答已采纳
我是新的flink用户，我有以下问题。使用flink将从关系数据库中提取的相关数据传输到HBase中。我用多个ExecutionEnvironments (每个RDB表一个用于并行传输表行)在java上编写flink批处理应用程序，以便按表顺序地传输表(因为调用env.execute()是阻塞的)。
我像这样开始纱线会话
export YARN_CONF_DIR=/etc/hadoop/conf
export FLINK_HOME=/opt/flink-1.3.1
export FLINK_CONF_DIR=$FLINK_HOME/conf
$FLINK_HOME/bin/yarn-sessi
浏览 1提问于2018-01-16得票数 7
回答已采纳
我在CheckpointedFuntion中使用运算符状态，但是在初始化MapState时遇到了NullPointerException：
public void initializeState(FunctionInitializationContext context) throws Exception {
MapStateDescriptor<Long, Long> descriptor
= new MapStateDescriptor<>(
"state",
TypeInfor
浏览 0提问于2018-03-18得票数 0
Spark有mapPartition API，而Flink DataStream API没有。有谁能帮助解释原因吗？我想要做的是在Flink上实现一个类似于星火reduceByKey的API。
浏览 2提问于2015-10-28得票数 7
回答已采纳
我正在为流分析方案评估Flink，并且还没有找到关于如何实现我们今天在遗留系统中所做的一种ETL设置的足够信息。
一个非常常见的场景是，我们使用了关键的、慢吞吐量的元数据流，我们希望使用这些数据流在高吞吐量数据流上进行丰富，如下所示：
这就引出了两个关于Flink的问题:如何通过缓慢更新时间窗口重叠但不相等的流来丰富快速移动的流(元数据可以存活数日，而数据可以存活几分钟)？如何有效地将多个(最多10个)数据流与Flink连接起来，比如一个数据流和9个不同的富集流？
我知道我可以使用非窗口的外部ETL缓存来实现我的ETL场景，例如使用Redis (这是我们今天使用的)，但我想看看Fli
浏览 3提问于2017-11-21得票数 1
回答已采纳
我有两个来自卡夫卡的事件源，一个来自user主题，另一个来自order主题，这两个主题的事件在我的代码中被创建为Flink动态表。
我有以下简单的连接查询，希望计算每个用户所放置的订单数。
select user_id,count(order_id) from user join order on user.user_id = order.user_id，
我有以下问题：
假设应用程序开始运行时，用户和订单中没有数据。
当Flink连接器从Kafka主题获得新数据时，会立即运行查询任务吗？查询中将包括多少个事件？我一次从卡夫卡控制台写了1000个用户和订单。。
如果查询(连接查询或复杂查询)需
浏览 6提问于2022-07-24得票数 0
回答已采纳
我试图在AWS EMR上部署我的flink作业(版本5.15和Flink 1.4.2)。然而，我无法从我的流中得到任何输出。我试着创造一个简单的工作：
object StreamingJob1 {
def main(args: Array[String]) {
val path = args(0)
val file_input_format = new TextInputFormat(
new org.apache.flink.core.fs.Path(path))
file_input_format.setFil
浏览 2提问于2018-07-19得票数 0
回答已采纳
我正在开发一个POC，将现有的Flink应用程序/拓扑转换为使用KafkaStreams。我的问题是关于部署。
具体来说，在Flink中，可以向flink安装添加"Worker节点“，然后为拓扑添加更多的并行化，以跟上不断增长的数据速率。
如何随着数据速率的增加而增加KStreams容量？KStreams会自动处理这个问题吗？我是否启动更多的进程(ala微服务)？
还是我错过了这里的大局？
浏览 4提问于2016-08-17得票数 3
回答已采纳
1回答
我有一个定制的flink Source，还有一个SerializableTimestampAssigner，它将事件时间戳分配给源发出的记录。由于底层数据存储的性质，源可能会发出无序的记录，但是在批处理模式下，我希望在任何操作符处理这些记录之前，Flink会根据事件时间戳对这些记录进行排序。
摘自Flink 的执行模式：
在批处理模式中，在预先知道输入数据集的情况下，不需要这样的启发式，因为至少可以按时间戳对元素进行排序，以便按时间顺序处理元素。
然而，情况似乎并非如此。如果我用时间戳分配器在Source (StreamExecutionEnvironment.fromSource)中
浏览 1提问于2022-04-22得票数 1
回答已采纳
我使用的是运行在Docker上的Flink 1.2.1，任务管理器作为Docker群的一部分分布在不同的am上。
使用Flink Web UI上传Apache光束应用程序并尝试在作业提交点设置并行度不起作用。两者都不会使用Flink CLI提交作业。
看起来在客户机级并行度并没有被接受，它最终默认为1。
当我以编程方式在Apache Beam代码中设置并行度时，它可以工作：flinkPipelineOptions.setParallelism(4);
我怀疑问题的根源可能在org.apache.beam.runners.flink.DefaultParallelismFactory类中，因为它
浏览 2提问于2017-10-23得票数 0
我正在尝试写案例类序列化到AVRO到HDFS使用RollingSink在Flink。为了让avro文件可以被HDFS反序列化，我使用了包装FSDataOutputStream的DataFileWriter。当我试图在DataFileWriter和FSDataOutputStream之间同步以便关闭HDFS上的数据文件时，抛出异常，实际上我每隔一个文件就会得到数据。有没有办法在Flink writer实现中同步fs流和Avro编写器？
我尝试过使用DataFileWriter close()、flush()、sync()、fsync()，但都失败了。Sync方法似乎执行得最好。我也尝试过在wri
浏览 1提问于2016-03-24得票数 0
我想使用Apache来处理应用程序中的事件。我在独立JVM上的测试运行得相当好，尽管flink是一个非常大的依赖项。
我还试图让它在OSGi中运行，但现在放弃了，因为有许多依赖项。
所以我的问题是:我能做多小的Flink。目前，我尝试使用maven依赖于flink流-java。
不幸的是，这取决于或嵌入(只列出有问题的)：
flink-阴影-hadoop2 2
克里奥
动物园管理员
奈蒂
防波堤
apache http客户端
apache http核心
斯卡拉
阿克卡
杰克逊
它还看起来像几个jars一次又一次地嵌入相同的库。就像一些谷歌libs
浏览 4提问于2015-12-18得票数 1
我计划动态缩放一个Flink应用程序。该应用程序使用kafka-flink连接器消耗来自Kafka的事件。
因为应用程序的“热身”只需几分钟(缓存.)而改变并行级别涉及重新启动，我更喜欢提交(放大)或交替杀死(缩小)任务，而不是改变并行级别。
我想知道，从性能、逻辑和执行计划来看，这种方法和Flink内置的并行执行有什么不同吗？
换句话说，10个相同的Flink任务与一个并行级别= 10的任务( env.setParallelism( 10 ) )有什么区别？
浏览 1提问于2019-09-25得票数 0
回答已采纳
1回答
我有两个来自两个Kafka主题的流表，我想加入这些流并对所连接的数据执行聚合功能。需要使用滑动窗口连接流。在连接和窗口数据时，我将得到一个错误Rowtime attributes must not be in the input rows of a regular join. As a workaround you can cast the time attributes of input tables to TIMESTAMP before.。
下面是代码片段
select cep.payload['id'] , ep.payload['id'] ,
浏览 25提问于2022-05-24得票数 0
2回答
如何在Flink转换中更新配置？
、、、、
给定一个Flink流作业，该作业将map()操作应用于流。
此map()操作从某些属性读取其配置，并相应地映射数据。例如，配置指定读取属性"input"，并使用不同的属性名"output“将其写入流。这个已经很好了。
现在，配置发生了变化，例如，转换将为输出使用不同的属性名。
因此，我正在寻找一种方式，让所有的Flink任务重新阅读一个新的配置在运行时。
有没有可能
将KafkaSourcewait挂起，直到管道耗尽(flush)trigger集群中的所有任务，以重新读取配置文件(coordinated)resume (coordinated)resume KafkaS
浏览 13提问于2022-10-11得票数 0
正在尝试在Amazon EMR Flink(5.21.0)上运行Flink(v1.7.0)作业。我得到了异常 java.lang.NoSuchMethodError: scala.Product.$init$(Lscala/Product;)V 异常看起来像是SCALA版本问题。我发现flink库是SCALA 2.11，而我的工作是使用2.12构建的。Flink 1.7.0支持SCALA 2.12。 问题是，如何让Amazon EMR支持SCALA 2.12而不是2.11？或者，除了我的代码回到2.11之外，还有更好的解决方案吗？
浏览 12提问于2019-03-01得票数 0
无法使用StreamingFileSink并以压缩方式存储传入事件。 我正在尝试使用StreamingFileSink向S3写入无界事件流。在这个过程中，我想压缩数据，以便更好地利用可用的存储大小。 我写了一个压缩字符串写入器，通过从flink借用SequenceFileWriterFactory的一些代码。它会失败，并出现我在下面描述的异常。 如果我尝试使用BucketingSink，它会工作得很好。我使用BucketingSink编写压缩字符串，如下所示。再一次，我从其他的拉取请求中借用了这段代码。 import org.apache.flink.streaming.connectors
浏览 58提问于2019-05-16得票数 2
2回答
我在流的世界里是个新手，在我的第一次尝试中我遇到了一些问题。 更具体地说，我正在尝试使用Flink在滑动窗口中实现计数和groupBy功能。 我在一个普通的DateStream中做过，但是我不能让它在WindowedStream中工作。 你对我该怎么做有什么建议吗？ val parsedStream: DataStream[(String, Response)] = stream
.mapWith(_.decodeOption[Response])
.filter(_.isDefined)
.map { record =>
(
浏览 18提问于2019-05-02得票数 1
我使用的是使用Java的Flink表API，其中我想将DataSet转换为DataStream .以下是我的代码：
TableEnvironment tableEnvironment=new TableEnvironment();
Table tab1=table.where("related_value < 2014").select("related_value,ref_id");
DataSet<MyClass>ds2=tableEnvironment.toDataSet(tab1, MyClass.class);
DataStream&
浏览 4提问于2016-05-13得票数 4
我需要在K8上部署Flink应用程序的帮助 我们有3个源，将以SQL查询的形式发送触发条件。总查询数约为3-6k，有效地增加了flink实例的负载。我试着执行，但它非常慢，并且需要很多时间才能开始。 由于大量的查询，我们决定为每个源创建多个flink应用实例。因此，一个flink实例实际上只能执行~1-2K个查询。 示例: sql查询源为A、B、C Flink实例： App A -->将只负责处理源A查询 App B -->将只负责处理源B查询 App C -->将只负责处理源代码C查询 我想在Kubernetes上部署这些实例 问题： a)可以使用迷你集群(内置)部署独立
浏览 31提问于2020-08-24得票数 1
1回答
理想情况下，我想运行一些实验来测量我的应用程序的输入使用者的压力(即，如果我的应用程序落后于输入数据主题，并且消息到达的速度快于它们的处理速度)。有人告诉我这很常见，但我不知道该怎么做 我正在阅读1.9版本的Metrics文档，从我得到的内容来看，我必须配置conf/flink-conf.yaml (在独立模式下)，假设jmx报告器是这样的： metrics.reporter.jmx.factory.class: org.apache.flink.metrics.jmx.JMXReporterFactory
metrics.reporter.jmx.port: 8789 那么，我是否应该使用
浏览 51提问于2020-06-15得票数 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
回答已采纳
我试图使用Table在Flink应用程序中利用s3a文件系统。我在EMR6.1(预装有Flink 1.11.0)。
作为应用程序的一部分，我正在s3位置上创建一个表。当我运行这个程序时，我会得到以下错误：
org.apache.flink.core.fs.UnsupportedFileSystemSchemeException
指示s3://或s3a:// URI都不支持任何文件系统。
每一个，如果我在电子病历，我不需要配置任何文件系统作为一个插件。
我反复检查了我的应用程序中没有包含fs-s3-hadoop依赖项。
奇怪的是，当我按照上面的说明将文件系统安装为一个插件时，我可以将它用于s3:
浏览 2提问于2021-09-01得票数 0
回答已采纳
我正在运行flink 1.6.1的单节点flink集群上运行字数统计示例
并不断得到这个错误。
我使用的是beam版本2.8.0。
错误
jobmanager_1 | 2018-10-28 07:20:41,401 ERROR org.apache.flink.runtime.webmonitor.handlers.JarRunHandler - Exception occurred in REST handler.
jobmanager_1 | org.apache.flink.runtime.rest.handler.RestHandlerException: org.apa
浏览 0提问于2018-10-28得票数 1
我在线程"main“中得到了一个异常:类org.apache.flink.state.api.runtime.SavepointLoader试图访问受保护的方法org.apache.flink.runtime.state.filesystem.AbstractFsCheckpointStorage.resolveCheckpointPointer(Ljava/lang/String;)Lorg/apache/flink/runtime/state/CompletedCheckpointStorageLocation；(org.apache.flink.state.api.runtime
浏览 0提问于2020-01-20得票数 1
回答已采纳
1回答
我正在创建2个flink sql表，1个用于CSV文件系统，另一个用于kafka。其目标是持续监视文件系统文件夹，并将新的csv文件记录推送到kafka主题。但是我在下面写的查询，推送csv文件记录一次，flink作业进入“完成”模式，任何新的文件都不会被处理。请有人告诉我如何创建flink sql连续流与源和csv文件系统和目标为Kafka。
创建源表的Flink SQL
CREATE TABLE son_hsb_source_filesystem_csv_bulk(
file_name STRING,
start_time STRING,
oss_cell_id B
浏览 1提问于2021-03-24得票数 0
回答已采纳
1回答
查看Flink的文档和书籍，我对时间戳有疑问:如果流设置为事件-时间模式，这意味着时间戳具有进入Flink之前的源时间(甚至在通过消息队列(可能是Kafka)之前)，为什么Flink将时间戳作为元数据附加到记录？幻灯片3根据时间戳的用途列出了不同类型的时间戳：
如果时间戳位于事件内部，为什么要将该值传递给记录的元数据？另外，事件和记录之间的确切区别是什么？
浏览 5提问于2021-02-22得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
