我想使用scala for Flink创建我自己的Sink，为此我需要扩展接口SinkFunction。 但是我不能覆盖下面的调用方法。我是Scala的新手，有没有什么可以弥补我缺失的东西呢？我在scala中看到的所有examples都使用
浏览 48提问于2019-12-20得票数 0
回答已采纳
关于Flink中的新sourceSinks接口，我有一个问题。我目前实现了一个新的自定义DynamicTableSinkFactory、DynamicTableSink、SinkFunction和OutputFormat。我以JDBC连接器为例，使用Scala。
输入到接收器的所有数据都具有Row类型。该框架提供运行时转换器，以便接收器仍然可以在公共数据结构上工作
浏览 5提问于2021-01-15得票数 0
回答已采纳
2回答
我遇到了几个可序列化的异常()，我在Flink的internet和文档上做了一些搜索；有一些著名的解决方案，比如瞬态、扩展序列化等等。每次异常的来源都非常清楚，但在我的例子中，我无法找到它到底在哪里没有序列化。class executor ( val sink: SinkFunction[List[
浏览 1提问于2017-12-22得票数 5
我是flink的新手，不知道这是正确的方法还是愚蠢的事情，我有一个字符串数据类型的数据流，我正在尝试将数据流中的数据捕获到一个列表中，我正在尝试类似下面的方法 } }
我
浏览 10提问于2020-10-09得票数 0
我对Apache Flink流API有一个问题。@RequiredArgsConstructor
public class EventDataConsumer extends RichSinkFunction<EventData，将一个方法引用传
浏览 4提问于2017-08-24得票数 0
回答已采纳
如果我的理解是错误的，我将向flink道歉，我正在构建一个数据流应用程序，并且该流包含多个数据流，这些数据流检查传入的DataStream中是否存在所需的字段。我试图模拟在一个DataStream中是否发生任何异常，其他数据流不应该受到影响，因为我在其中一个流中显式抛出一个异常。在下面的示例中，为了简单起见，我使用windows
浏览 1提问于2020-10-27得票数 0
回答已采纳
1回答
早些时候我问过Flink的一个简单的hello world example。这给了我一些很好的例子！ 然而，我想要一个更“流”的例子，我们每秒都会生成一个输入值。这在理想情况下是随机的，但即使每次都是相同的值也是可以的。 其目标是获得一个没有/最少外部接触的“移动”流。 因此我的问题是： 如何在没有外部依赖的情况下显示Flink实际流数据？我</
浏览 22提问于2020-04-27得票数 0
回答已采纳
我在集群上使用Flink。在提交任务时，我得到了以下例外：Kafka版本: 2.2.2object batchProcess { valbatch2feature_userId
浏览 5提问于2022-06-26得票数 0
我们正在使用火花流-卡夫卡-0-8接收器。我们无法通过增加numPartitions来增加消耗事件的数量。增加numPartitions似乎不影响性能。KafkaUtils.createStream方法有一个topic_name到numPartitions映射，而每个分区应该在自己的线程中使用。(topic -> 1),我希望使用scala.collection.
浏览 4提问于2017-12-26得票数 0
我有一个有一个超类数据模型的n个Java类。类的列表是Scala方法的输入参数，我想在该方法中创建resultStreams，并必须从process方法创建调用。你能写一下怎么解决吗？我试图在方法调用中使用_ <：SpecificRecordBase，SpecificRecordBase，但结果相同。extends SpecificRecordBase> getClas
浏览 2提问于2020-02-06得票数 0
回答已采纳
我阅读了使用RichSinkFunction比直接调用DB方法的优点。因此，我决定写我自己的RichSinkFunction。我想在一个单独的类下调用这个方法，这个类应该传递代码中提到的3个参数。记录采用JSON格式。我可以通过解析和获取属性来管理这一点。但是如何将它传递给invoke方法，以及如何在整个类中传递session对象。而且，这是一种正确的方法吗?因为
浏览 4提问于2022-01-05得票数 1
如何在Flink流中实现反馈循环。换句话说，我如何才能编写一个也是SinkFunction的SourceFunction？DataStream<Output> outputStream = controlSignal.flapMap(FlatMapFunction)
现在，我希望将outputStream发送到接收器，并根据一些逻辑将数据发送到ControlStream。一种方法是让ControlStream从卡夫卡和我可以从卡夫
浏览 0提问于2018-05-17得票数 1
回答已采纳
我计划在首选的位置发射接收器。我有一个包含10个工作节点的spark独立集群。我想根据我的规范在所有工作节点中启动4个接收器。为此，我在客户接收器中使用以下代码。@Override scala.Option<String&g
浏览 3提问于2016-12-23得票数 0
我正在尝试为ArangoDB编写一个自定义的ArangoDB。我使用CassandraOutputFormat.java作为灵感，交换了相关的驱动程序和调用等等。我用Java编写了这篇文章，因为我找不到OutputFormat的Scala示例。因此，该类的定义如下：
public class ArangoOutputFormat<OUT extends Tuple> extends RichOutputF
浏览 3提问于2016-09-09得票数 1
回答已采纳
2回答
我每隔5分钟就会从ElasticSearch读取数据到Spark。因此，每5分钟就会有一个RDD。我希望在这些RDDs的基础上构建一个DStream，这样我就可以得到过去1天，过去1小时，过去5分钟等数据的报告。为了构建DStream，我在考虑创建我自己的接收器，但是spark的官方文档只提供了使用scala或java来实现<e
浏览 0提问于2016-03-15得票数 0
2回答
最近，我开始学习流处理，并尝试使用Apache。我试图编写一个作业，从Kafka主题读取事件，可能执行一些无状态链式转换，并对另一个应用程序进行REST调用，以发布每个转换后的事件。RestSinkFunction<>()); //Custom sink function implementing org.apache.flink.streaming.api.functions.sink.SinkFunctionenv.execute("Confused Job&
浏览 2提问于2019-12-10得票数 0
我跟随从Blob中的平面文件加载Azure中的数据。
此方法需要一个C# .NET框架控制台应用程序来创建/管理Blob、AzureSQL接收器和dataset。唯一的问题是，对于大的、杂乱的数据文件，我不可避免地会遇到一些行，这些行包含一个额外的分隔符，或者以某种方式出现格式错误。数据太大，无法在上传到blob之前在本地进行清理。有一个如何使用
浏览 1提问于2020-04-04得票数 1
我在一个纱线集群的所有执行器之间均匀分配流接收器时遇到了问题。我有一个包含8个执行器的yarn集群，我创建了8个流自定义接收器，spark应该为每个执行器启动一个接收器。然而，这并不总是发生，有时所有的接收器都是在同一个执行器上启动的(这里是jira的bug：)。所以我的想法是运行一个虚拟作业，获取该作业中涉及
浏览 1提问于2015-09-22得票数 1
1回答
所以我想做一个自己的Functor来解决类型化问题。然而，我遇到了一堆问题。最终的目标是能够说出像Seq(1,2,3).fmap(2*_)这样的东西。我想出了这个trait，我在谷歌上的所有搜索都(差不多)都同意： def fmap[A, B](r: F[A], f: A => B): F[B]我假设没有简单的方法将但我在
浏览 0提问于2017-10-10得票数 1
我已经使用完全模式将聚合应用于流数据格式。为了在本地保存数据格式，我实现了foreach接收器。我能够以文本形式保存数据。但我需要用Parquet格式保存它。columnName))我怎样才能做到这一点
浏览 0提问于2017-09-26得票数 5
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
腾讯会议
云直播
对象存储活动推荐
腾讯云开发者
