时间戳

首页 > 解决方案 > 动态调用 Flink 算子

apache-flink - 动态调用 Flink 算子

问题描述

我最近开始学习流处理，并且正在尝试使用 Apache Flink。我正在尝试编写一个从 Kafka 主题读取事件的作业，可能执行一些无状态链式转换，并对另一个应用程序进行 REST 调用以发布每个转换后的事件。例如，我的主要方法可能如下所示 -

public class KafkaSourceToRestSinkJob {
    public static void main(String[] args) {
        String configPath = args[0];
        //Read configuration for the job (like kafka properties, rest uri for sink, possibly operators to invoke)
        ...
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> dataStream = env.addSource(new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), kafkaProps));
        dataStream.addSink(new RestSinkFunction<>()); //Custom sink function implementing org.apache.flink.streaming.api.functions.sink.SinkFunction
        //Chain some operators depending on some parameters in the config file
        ...
        env.execute("Confused Job");
    }
}

我的目标是为具有相同类型的源和接收器的多个作业提供一个通用的 jar 工件。如果我需要一项工作来执行转换 A、B 和 C（实现将存在于 jar 中），我可以在配置文件中指定它们并将路径传递给程序 args 中的文件。

现在这是我的问题 -

  1. 是否可以动态调用运算符？
  2. 我知道在接收器中进行 REST 调用可能会导致一些不必要的延迟，但在我的应用程序中，这是可以容忍的。我也不在乎反应。牢记这一点，我是否有理由避免使用 REST 接收器？
  3. 总的来说，我是不是大错特错了？

谢谢！

标签： apache-flinkflink-streamingstream-processing

解决方案

我可能会看看 Flink SQL。您可以定义公共源/接收器，然后只需将SQL 查询传递给 Flink。

过去我使用 Spark SQL 进行了类似的设置，并且效果很好。您不需要发明自己的规范语言，而且更容易理解。

推荐阅读