2回答
现在，我已经编写了一个使用者来阅读该主题并构建一个GenericRecord数据流。现在，我不能将此流汇到parquet格式的hdfs/s3中，因为此流包含不同类型的模式记录。因此，我通过应用过滤器并创建不同的流，然后分别下沉每个流，为每种类型过滤不同的记录。;import org.apache.flink.core.fs.Path;
浏览 69提问于2020-02-21得票数 1
回答已采纳
1回答
流在Hadoop中的应用
、、、、
在这个例子中，我确实看到他们创建了一个单独的应用程序(类似于流应用程序)，该应用程序生成和消费数据，并在数据上应用模式匹配。他们现在还没有在.Till(如卡夫卡)之间放置一个流层，单应用程序就足以满足这一目的，这使得它非常优化。我有几个问题我没有得到回答：
如果像卡夫
浏览 2提问于2016-06-13得票数 0
FLINK流：，我有来自kafkaconsumer的DataStreamString，这是JSON .addSource(new FlinkKafkaConsumer[topic", new SimpleStringSchema(), properties))
val input: DataStream[GenericRec
浏览 16提问于2020-07-24得票数 0
KStream<String, KafkaDataType> eventStream = builder.stream(STREAM_TOPIC);
我的KafkaDataType是一个由关联的.avsc文件自动生成的我的理解是KafkaDataType必须是预先定义的.但是，是否有允许动态或通用KafkaDataType的现有方法？如果是这样的话，最好使用一个示例代码块。我们的目标是让KafkaDataType成为一个通用数据类型，这样就可以将具有不同avro的不同的Kafka流交换进来和输出，并由Java代码进行处理。目前，
浏览 5提问于2021-11-03得票数 0
回答已采纳
我在我的flink stream工作中看到了一个奇怪的行为。> enrichDataStream = dataStream.map(new MapFunction<GenericRecord, GenericRecord>() { @Override
public GenericRecord map(GenericRecord现在我需要从我的avro
浏览 379提问于2020-05-24得票数 2
尝试实现一个用于读取Kafka流和聚合会话的Flink作业，由于某种原因，getResult()没有被调用。source.keyBy(new KeySelector<GenericRecord, String>() { return record.get("id").toStringNullable
浏览 76提问于2019-11-28得票数 1
我们有两种信息要发送到Flink
import org.apache.flink.streaming.connectors.fs.RollingSink;
}
但我观察到的是，
浏览 3提问于2016-05-06得票数 5
1回答
由于Apache的1.15版本，您可以使用压缩功能将多个文件合并到一个文件中。此外，我们找不到任何的例子，压缩Parquet或其他散装格式。
浏览 49提问于2022-06-22得票数 2
我正在构建一个Flink流应用程序，并且更倾向于使用事件时间，因为它将确保所有已设置的计时器在历史数据发生故障或重播时决定性地触发。事件时间的问题是，只有当事件发生时，时间才会向前移动。我们建议的解决方案是使用AWS lambda函数，该函数计划每隔X分钟运行一次，该函数将虚拟事件输出到Flink读取的Kinesis流中，从而强制生成一个水印以向前推进时间。这些文档让我相信，Flink并行地从一个源读取，其中每个并行读取操作符生成自己的水印，然后一个下游操作符，比如一个
浏览 2提问于2019-12-06得票数 0
回答已采纳
我有一个Flink程序，它接受两个流，即数据/传感器读数流和警报规则流。我正在广播规则流，并将其连接到数据流以生成动态警报。我已将时间戳&水印分配给数据流，并按原样传递规则流(因为只有在看到新的附加规则/更新时，规则流才会有记录)。但是没有产生警报。当两个流，即一个具有时间戳和水印的流和一个有规则的流(广播)按照规则<
浏览 0提问于2019-08-21得票数 2
回答已采纳
在处理每个数据流的多个数据流和多个操作符时，我试图理解Flink的运行时方面。用例:单个flink作业中的n个数据流(每个数据流代表一个设备--具有不同的时间延迟)，每个数据流被分成两个流，其中一个流被分成几个CEP操作符，另一个流为一个进程函数。问题：
在作业开始时，可以在运行时动态创建数据流
浏览 2提问于2017-11-09得票数 1
1回答
对我来说，很难理解Flink中的流表。我可以理解Hive，将一个固定的静态数据文件映射到一个“表”，但是如何体现一个建立在流数据之上的表呢？例如，每1秒就有5个具有相同结构的事件被发送到Kafka流：{"num":2, "value": "b"}
....构建在它们上的动态表是什么样子的？Flink将它们全部消耗，并存
浏览 2提问于2020-10-29得票数 0
回答已采纳
1回答
我正在尝试根据数据库中的某些规则从一个原始数据流中创建新的数据流。启动Flink作业后，如果我需要根据数据库中更新的条目创建新的数据流，我如何才能做到这一点。因此，这里我不想重新启动Flink作业，而是在运行时动态创建新的数据流，并将其添加到执行DAG中。
浏览 4提问于2016-06-27得票数 2
1回答
我有一个关于在动态流中分片数据的问题。在将用户数据发送到我的动态流时，我希望使用一个随机分区键，以便碎片中的数据分布均匀。为了简化这个问题，我想通过在我的Flink应用程序中键入一个userId来聚合用户数据。我的问题是:如果碎片是随机分区的，以便将一个userId的数据分散到多个Kinesis中，那么Flink能否处理多个碎片的读取，然后重新分发数据，以便将单个userId的所有数据都流到同一个聚合器任务中或者，在Flink使用之前，我是否需要按用户id分割这个运动<
浏览 3提问于2020-02-14得票数 0
回答已采纳
从Flink 1.5发布公告中，我们知道Flink现在支持“广播状态”，并描述了“广播状态为Flink的CEP库的”动态模式“特性的实现打开了障碍。”这是否意味着目前我们可以使用“广播状态”来实现“动态模式”而不需要Flink CEP？另外，我不知道在有或没有广播状态的情况下实现Flink CEP的“动态模式”有什么区别？=============
测试广播数据流的更新.使用键控-datastream的操作员广播()
浏览 1提问于2018-05-26得票数 0
回答已采纳
下面是我使用flink sql API连接两个表的代码 tEnv.createTemporaryView("A", streamA,"speed_sum,cnt,window_start_time,window_end_timetrue,6799.388891220093,310,1635333650000,1635333660000) // want to save 如您所见，toRetractStream应用编程接口将生成三条记录
浏览 211提问于2021-10-27得票数 1
经过一段时间的思考和重组我的计划，我能够推动我的实现到目前为止： public static FlinkKafkaProducer<Tuple2<GenericRecord,GenericRecord,GenericRecord>>("flink_output",
import o
浏览 11提问于2020-01-30得票数 0
回答已采纳
我已经写了一个示例代码，在从卡夫卡读取GenericRecord后，可以将流写入parquet格式 Properties config = new Properties(); Schema schema = new Schema.Parser().parse(file);
浏览 40提问于2020-01-19得票数 0
1回答
我使用jdbc source connector从oracle获取数据，并将( avro格式的键和值(模式注册表))推到kafka主题。:"subject": "testtopicA-key","id": 1123,}
我有一个卡夫卡流听这个话题当我启动流时，我开始获得ClassCa
浏览 4提问于2022-03-31得票数 0
我已经编写了flink作业来使用数据集和数据流api。我已经从相同的main()方法启动了这两个程序，但现在当我将作业提交到flink Dashboard时。只有dataset api程序正在运行，仪表板显示作业已完成，流未被触发，flink仪表板也未显示有关数据流执行的任何内容。但是，当我从eclipse运行时，数据集和数据流api程序都在运行。是否有其他方法可以将作业提交给flink作业管理器以同时完成这两项工作
浏览 16提问于2017-07-12得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
腾讯会议活动推荐
腾讯云开发者
