动态调用Flink运算符

Question

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

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. 总的来说，我犯错了吗？

谢谢！

Answer 1

您不能动态地修改作业图的拓扑，但是，例如，您可以实现一个flatmap运算符，该运算符可以动态加载一个类（在配置中指定），然后使用它来转换事件流。

至于REST接收器，如果您需要端到端仅保证一次语义，那么您就需要使用Flink的检查点小心地使接收器适合。 FlinkKafkaConsumer通过倒退和重播自上一个检查点以来的事件来处理恢复。如果不小心，将导致在恢复期间将重复的结果推送到REST接收器。如果REST接收器仅在外部系统上执行幂等更新，那么就可以了，但是否则，您需要使REST接收器有状态且具有事务性。

Answer 2

我可能会看一下Flink SQL。您可以定义通用的源/接收器，然后只需传递SQL query to Flink。

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