集群中的Apache Flink流不会与工人拆分作业

Question

我的目标是使用Kafka作为源设置一个高吞吐量集群

我在主服务器和辅助服务器上设置了一个2节点集群，配置如下。

flink-conf.yaml大师

jobmanager.rpc.address: <MASTER_IP_ADDR> #localhost

jobmanager.rpc.port: 6123

jobmanager.heap.mb: 256

taskmanager.heap.mb: 512

taskmanager.numberOfTaskSlots: 50

parallelism.default: 100

Worker flink-conf.yaml

jobmanager.rpc.address: <MASTER_IP_ADDR> #localhost

jobmanager.rpc.port: 6123

jobmanager.heap.mb: 512 #256

taskmanager.heap.mb: 1024 #512

taskmanager.numberOfTaskSlots: 50

parallelism.default: 100

主节点上的从属文件如下所示：

<WORKER_IP_ADDR>
localhost

两个节点上的 flink 设置位于具有相同名称的文件夹中。我通过运行

bin/start-cluster-streaming.sh

这将启动Worker节点上的任务管理器。

我的输入源是Kafka。以下是片段。

final StreamExecutionEnvironment env = 
    StreamExecutionEnvironment.getExecutionEnvironment();

DataStreamSource<String> stream = 
    env.addSource(
    new KafkaSource<String>(kafkaUrl,kafkaTopic, new SimpleStringSchema()));
stream.addSink(stringSinkFunction);

env.execute("Kafka stream");

这是我的水槽功能

public class MySink implements SinkFunction<String> {

    private static final long serialVersionUID = 1L;

    public void invoke(String arg0) throws Exception {
        processMessage(arg0);
        System.out.println("Processed Message");
    }
}

这是我的pom.xml.中的Flink依赖项

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-core</artifactId>
    <version>0.9.0</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>0.9.0</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>0.9.0</version>
</dependency>

然后，我在主服务器上用这个命令运行打包的jar

bin/flink run flink-test-jar-with-dependencies.jar

但是，当我将消息插入Kafka主题时，我能够在主节点上单独考虑来自Kafka话题的所有消息（通过我的SinkFunction实现的invoke方法中的调试消息）。

1. 为什么工作线程节点没有获取任务？
2. 我是否缺少某些配置？

Answer 1

在Flink中从Kafka源读取时，源任务的最大并行度受到给定Kafka主题的分区数的限制。Kafka分区是Flink中源任务可以使用的最小单位。如果分区多于源任务，则某些任务将消耗多个分区。

因此，为了向所有100个任务提供输入，您应该确保您的Kafka主题至少有100个分区。

如果您无法更改主题的分区数，那么也可以使用setParallelism方法以较低的并行度从Kafka开始读取。或者，您可以使用重新平衡方法，该方法将在前面操作的所有可用任务中洗牌数据。