我能够将静态行接收到数据库中： DataStream<Row> staticRows = environment.fromElements("value1", "value2")tableEnv.executeSql(myDDLAndSinkProperties);
inputTable.executeInsert("MYTABLE"); 但是像这样将一个无界的流映射</e
浏览 230提问于2021-10-16得票数 1
回答已采纳
我正在使用Flink SQL，下面的方案显示了我的源数据(属于某些Twitter数据)： `entities` ROW(STRING, `id` BIGINT )WITH (...);
浏览 15提问于2021-12-23得票数 2
回答已采纳
Added Depedency Pom Details :
<dependency> <artifactId>flink-clients_2.11</artifactId> <
浏览 0提问于2019-04-04得票数 2
回答已采纳
1回答
我们有包含(大约)20M行和50,000列的数据(静态)。这是一个稀疏数据，我们需要快速读取单个单元格值或给定列(该列的所有行)。输入是一个流数据，我们希望根据来自数据库的值(上面提到的值)聚合输入(最后一个x分钟)。 我们需要一些关于如何实现最低延迟的建议：1.Apache Flink被排除在考虑之外，因为在向堆栈中添加新框架时遇到了阻力，我们正在寻找更稳定的框架(因为这个问题只是项目的一部分)。
浏览 10提问于2019-03-20得票数 0
因此，我正在用Storm和Flink应用程序构建一个JAR，其中我记录消息如下：import org.slf4j.LoggerFactory;
LOG.info("...");然后，我将JAR传递给.../bin/storm和.../bin/flink脚本，一切都正常，但是日志级别被设置为IN
浏览 4提问于2019-10-11得票数 0
回答已采纳
我对Flink的表-api (1.13+)有一些问题.我有一个包含几个字段的POJO，其中之一是：我为这个字段使用以下声明来创建我的表： ...之后，当我试图使用DataStream方法将表转换为toRetractStreamMY_POJO时，我得到了以下错误：
未注册表的查询结果和接收器的线程“主”or
浏览 2提问于2022-01-19得票数 5
我想做一些DataSet转换，但是Flink中的最大列Tuple是Tuple25，包含25列。java.lang.IllegalArgumentException: The tuple type exceedsthe maximum supported arity
如果我不扩展Tuple类，我将这些行
浏览 0提问于2018-06-17得票数 2
回答已采纳
下面是一个转换的简化表示，我需要将它应用于一个大型复杂表。在这里，输入表是一个有两个数字列的查找。raw和SS有着多对一的关系.raw的许多值可以具有相同的SS值，但反之亦然。raw列强制使用字符，有三种类型的值来保留原始SS：的多对一关系。一个破折号“-”表示SS的值，没有对应的raw值(例如，SS = 70)
SS值的单个</em
浏览 1提问于2019-07-10得票数 1
回答已采纳
我正在构建一个流解析器，它要求单个表的初始内存状态估计为50 GB。~50 GB是将状态加载到Scala HashMapString字符串中时所使用的内存数量。对于一个表，Flink是否能够处理具有~50 GB的状态，并且随着时间的推移而增长？备注：
I不能将类型更改为任何更小的类型。状态被用作查找将一个字符串映射到另一个字符串<em
浏览 5提问于2019-10-28得票数 2
回答已采纳
我对flink和kafka流很陌生，所以请尝试通过给出解决方案代码来回答。我希望能够将它转换为包含所有键和值对的Map。map1.get("FC196")应该让我处于休眠状态，map1是包含键值对的映射
我面临的挑战是将代码中的第一个变量DataStreamObjectNode转换为键值对的映射。.addSource(new FlinkKafkaConsumer09("new", new Sim
浏览 6提问于2017-04-25得票数 0
回答已采纳
2回答
我正在使用Apache开发一个项目，并使用junit测试我的操作员。但是，我面临一个问题:由于并行性，flink将以“随机”行顺序写入其输出csv文件，因此我很难断言输出文件与使用Junit的预期输出文件相等。性能不是一个问题，因为我们谈论的是小文件(<100行)，而且只用于测试。
有什么简单的解决办法吗？
浏览 4提问于2015-04-08得票数 5
回答已采纳
3回答
Flink自定义分区函数
、、、
我在DataSet应用程序接口的Flink上使用Scala。我想跨节点对数据进行重新分区。Spark有一个功能，可以让用户使用给定的numberOfPartitions参数(link)重新划分数据，但我相信Flink不支持这种功能。因此，我想通过实现一个自定义分区函数来实现这一点。我的数据类型为DataSet(Double，SparseVector)，数据中有一行示例： (1.0 SparseVector((2024,1.0), (2025,1.0), (2030,1.0), (Doub
浏览 33提问于2019-01-15得票数 5
请您帮帮我--我正在尝试使用Apache Flink进行机器学习任务，外部集成/树形库(如XGBoost )，所以我的工作流程如下：
接收单个数据流，其中原子事件看起来像一个简单的向量event=(将这些提取的特征放在数据集/表列的10分钟窗口中，并在这样的10分钟数据上运行最终机器学习任务。简单地说，我需要对流中的同一个单个事件应用几个完全不同的map操作，然后在单
浏览 2提问于2017-10-08得票数 1
回答已采纳
1回答
在Flink作业中，我读取了Kafka流，并在将数据保存到数据库之前应用了一些联接。Kafka主题包含两种类型的数据，因此我首先连接两个记录，创建单个行，并将其保存到数据库中。我需要最新的数据存储在数据库中。from tableA" + " on a.id=b.id, a.item=b.item");
最后，
浏览 10提问于2022-09-24得票数 0
回答已采纳
3回答
我试着通过Flink获取Kinesis的数据。我试着搜索它，但我找不到合适的答案。我的代码所做的是从Kinesis获取数据，解压缩gzip，将其转换为字符串，然后使用objectMapper.readvalue到我的POJO for java。
浏览 4提问于2021-09-24得票数 2
1回答
单源多汇v/s平面图
、、、
我在Flink上使用Kinesis Data Analytics进行流处理。 .name("rawData")
.addSink(createS3SinkFromStaticConfig("raw_data我看到了一个很大的CPU峰值(与只有一个接收器的CPU相比)。我看到
浏览 7提问于2021-01-29得票数 0
我有一个简单的Flink流处理应用程序(Flink版本1.13)。Flink应用程序从Kakfa读取，对记录进行有状态处理，然后将结果写回Kafka。在Kakfa中用于分区的键是记录的字符串字段(使用默认的kafka分区程序)。卡夫卡主题有24个分区。
映射类定义如下。它跟踪记录的状态。KeyGroupRange{startKeyGroup=96, endKeyGroup=127} does not contain
浏览 35提问于2022-06-15得票数 0
2回答
我有卡夫卡的阿夫罗格式的留言。它们必须转换为表，然后使用SQL进行选择，然后转换为流，最后转换为接收器。有多个具有不同Avro模式的Kafka主题，因此需要动态表。这是我正在使用的代码StreamTableEnvironment tableEnv = StreamTableEnvironment.create这门课的代码。当我打印表描述时，我看到项的类型为ANY。STRING | true | |
浏览 0提问于2020-12-29得票数 2
模型在测试集上的准确率为99.33%，所以我知道它工作得很好。使用model.predict正确地对单个样本进行分类，但是当我在Excel版本中尝试相同的示例时，预测通常是不同的(而且不正确)。)
然后将数据保存为.xlsx格式，并将网格复制粘贴到Excel模型中。归一化图像是通过将每个单元格的值除以255个来创建的，然后是0填充。， ReLU激活是使用最大(0，单元格)对每个单元进行的卷积特征映射，为5个过滤器中<
浏览 1提问于2020-06-30得票数 2
我试图在flink中实现一个批处理数据处理管道，在这里我需要
向下游系统发布包含一些元数据的通知，这些系统可以读取elasticsearch问题1:我没有找到将数据发布到elasticsearch的任何简单方法，flink dataset API中没有提供任何开箱即用的输出函数来将数据发布到elasticsearch，我只能找到写入文件或实现自定义函数<e
浏览 1提问于2019-10-10得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
腾讯会议
云直播
对象存储活动推荐
腾讯云开发者
