apache-flink

0热度

1回答

我有两个Apache Flink集群：生产中的1.1.3和分段中的1.3.2。我有兴趣将指标发送到Graphite服务器，因此我按照https://ci.apache.org/projects/flink/flink-docs-release-1.3/monitoring/metrics.html中的说明进行设置。我得到它在我的1.1.3群集，但不是在1.3.2。我加入弗林克lib目录下的ja

0热度

2回答

flink 1.2下沉kafka流的错误

我所做的是读取来自kafka的消息以json格式。例如。 {"a":1,"b":2} 然后我应用的滤波器此消息，以确保对应于a的值是1，b的值是2。最后，我想以输出结果流至下游卡夫卡。但是，我不知道编译器为什么说类型不匹配。我的代码如下： val kafkaConsumer = new FlinkKafkaConsumer010( params.getRequired("input-topi

2热度

1回答

我使用正确的框架？

我是scala/flink/spark的新手，会有几个问题。现在scala正在使用flink。数据流的总体思路是这样的： CSV文件 - >弗林克 - > elastic->弗林克（过程数据） - > MongoDB的 - >的Tableau 有其分号日志文件的数量巨大分隔。我想将这些文件写入elasticsearch作为我的数据库。（这已经有效）现在需要各种分析（f.e.一致性报告/生

0热度

1回答

在下沉kafka流时看不到消息，无法在flink 1.2中看到打印消息

我的目标是使用kafka读取json格式的字符串，对字符串进行过滤，然后将消息下沉（仍在json字符串中格式）。出于测试目的，我的输入字符串信息是这样的： {"a":1,"b":2} 而且我实现的代码是： def main(args: Array[String]): Unit = { // parse input arguments val params = ParameterTool

2热度

1回答

阿帕奇弗林克：NullPointerException异常引起TupleSerializer

当我执行我的弗林克应用它给了我这个NullPointerException： 2017-08-08 13:21:57,690 INFO com.datastax.driver.core.Cluster - New Cassandra host /127.0.0.1:9042 added 2017-08-08 13:22:02,427 INFO org.apache.flink.runtime.t

0热度

1回答

Flink 1.3.1，Elasticsearch 2.4.6，RuntimeException：Elasticsearch客户端未连接到任何Elasticsearch节点

试图在Flink培训/ exersizes上取得进展。有如下issue。重新安装Elasticsearch，现在版本为：2.4.6。 Elasticsearch似乎工作。我可以通过http://localhost:9200访问它。创建index和type mapping根据练习指令使用cURL命令。安装了Kibana：4.6.5，Kibana能够连接到Elasticsearch并解析我创建的

1热度

2回答

Apache Flink：MapPartition VS. FlatMap

我没有清楚地了解MapPartition。请有人解释一下Mappartition的用例以及它与FlatMap的不同之处？

0热度

1回答

如何在Flink 1.2中从Kafka中提取部分json格式的字符串

我的目标是使用kafka以json格式读取字符串，对字符串进行过滤，选择部分消息并下沉消息（仍以json字符串格式）。出于测试目的，我的输入字符串信息是这样的： {"a":1,"b":2，"c":"3"} 而且我实现的代码是： def main(args: Array[String]): Unit = { val inputProperties = new Properties() i

0热度

1回答

kafka-flink：根据flink作业的卡夫卡制作者错误

虽然我在做卡哇卡flink的概念验证，但我发现了以下情况：似乎kafka制作者的错误可能是由于flink端完成的工作量而发生的！以下是详细信息：我有样本文件样？EDR包含“实体”，“价值”，“时间戳” 我用值进行700'000行的〜。下面的命令来创建卡夫卡话题： ~/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --

0热度

1回答

Kubernetes Autoscaling - 触发事件

我试图让Flink集群运行在kubernetes之上。由于当前的Flink 1.3.1还不支持动态缩放，因此当使用kubernetes向上/向下缩放时，我想要触发我的自定义脚本停止并从保存点重新开始作业。在Kubernetes中是否有这样的功能，我可以在自动缩放中使用前/后事件触发点，以便在自动缩放完成其作业后运行自定义脚本？或者现在看来唯一的选择是编写自己的cron监控来使用kubectl规