问题标签 [flink-batch]

我正在使用 Apache Flink Streaming API 来处理数据文件，并且我对仅从最后一个窗口中获取结果感兴趣。有没有办法做到这一点？如果不可能，我想我可以过滤结果元组中第一个字段的最大值（长值）。

在上面的过滤转换中，它会通过元组的第一个字段的最大值进行过滤。有可能以某种方式做到这一点吗？

我正在努力将hdfs集成到flink。

Scala 二进制版本：2.12，

Flink（集群）版本：1.10.1

这是HADOOP_CONF_DIR；

hdfs的配置在这里；

这个配置和HADOOP_CONF_DIR在taskmanager里也是一样的。

pom.xml;

我正在尝试从 hdfs 获取镶木地板文件，我的示例代码在那里；

错误在这里；

当您在 lib 文件夹下看到 Hadoop uber jar 时，这很奇怪

这就是我提交工作的方式；

docker exec -it jobmanager env HADOOP_CONF_DIR=/tmp/hadoopconf flink run -C file:///opt/flink/lib/flink-shaded-hadoop-2-uber-2.7.5-10.0.jar -d /tmp/core -batch-1.0.0.jar

我也尝试使用 flink UI 推送作业，但结果是一样的。

任何帮助，将不胜感激。

谢谢。

有人可以解释如何将 Excel 数据加载到 Apache Flink 中吗？我在 API Doc 中看到了其他类型的格式，例如 txt、csv，但不是 Excel。

提前致谢。

我是 flink 新手，想了解如何使用 FLINK 运行我的用例：应用程序具有三个输入数据源 a) 历史数据 b) 从 kafka 获取所有实时事件 c) 获取将具有触发条件的控制事件

由于应用程序正在处理历史数据，所以我认为我将合并历史数据和实时数据，并在该流上创建一个表。

要触发事件，我们必须借助作为输入源且包含 where 子句的控制事件来编写 SQL 查询。

我的问题是构建 SQL 查询，因为数据在 Stream 中，当我执行类似的操作时

它不会引发序列化异常 Localexecutionenvironment

我正在从 Kafka 读取数据并尝试以 ORC 格式将其写入 HDFS 文件系统。我使用了他们官方网站上的以下链接参考。但我可以看到 Flink 为所有数据写入完全相同的内容并制作了这么多文件并且所有文件都可以 103KB

https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/connectors/streamfile_sink.html#orc-format

请在下面找到我的代码。

我想知道FLINK是否可以支持我的要求，我已经浏览了很多文章但不确定我的情况是否可以解决

案例：我有两个输入源。a)Event b)ControlSet 事件样本数据为：

如果您看到 event-1 和 event-2 在“数据”中都有不同的属性。所以考虑像数据是自由格式字段并且属性的名称可以相同/不同。

ControlSet 会给我们指令来执行触发器。例如触发条件可能是

如果这种场景可以在 flink 中运行，请帮助我，什么是最好的方法。我不认为 patternCEP 或 SQL 可以在这里提供帮助，并且不确定事件 dataStream 是否可以作为 JSON 对象并且可以像 JSON 路径一样查询。

我有多个 Kafka 主题（多租户），并且我根据主题数量多次运行同一个作业，每个作业都使用来自一个主题的消息。我已将文件系统配置为状态后端。

假设有 3 个作业正在运行。检查点如何在这里工作？所有 3 个作业是否都将检查点信息存储在同一路径中？如果任何作业失败，作业如何知道从哪里恢复检查点信息？我们过去在向 flink 集群提交作业时会给出作业名称。它与它有什么关系吗？一般来说，Flink 如何区分作业及其检查点信息以在作业失败或手动重新启动作业的情况下进行恢复（无论作业相同还是不同）？

案例1：如果工作失败会发生什么？

Case2：如果我们手动重启作业会发生什么？

谢谢

https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/hive/#connecting-to-hive 根据这个链接，Flink 需要本地 hive conf 文件夹路径，但我需要提交纱线上的 Flink 作业，因此 Flink 尝试在纱线容器中查找路径，例如/mnt/volume4/yarn/nm/usercache/akashkumar.patel/appcache/application_1594626939821_80078/container_e83_1594626939821_80078_01_000002/hdfs:/warehousestore/hive/warehouse/db/hive_conf/hive-site.xml

我们该如何处理？，我只需要为 hive 表创建一个分区。有没有办法为 hive-conf 文件夹提供 hdfs 文件夹位置？

解释：

假设我有 3 个事件[E1，E2，E3]，应该按计数触发，也应该按时间触发。我正在使用 countTrigger 仅触发 2 个事件（E1 和 E2），但未触发剩余的 E3 事件。

预期：E3 事件应在 5 秒后触发，但实际上它仅触发 E1 和 E2 事件

我有两个来源，一个是 Kafka 来源，一个是自定义来源，我需要制作一个睡眠自定义来源一小时，但我遇到了中断。

代码：

如何在 Kafka 源将继续其流时制作睡眠自定义源。为什么我得到线程中断异常？