Apache Beam 是一个基于批处理和流处理模型的分布式处理框架，可以轻松使用 Apache Kafka 作为数据源进行大规模的数据处理。对于从 Kafka 中消费数据，Beam 提供了两个 API：ReadFromKafka 和 KafkaConsume。

ReadFromKafka 是一个高级别的 Beam API，它从 Kafka 获取消息并将其转换为 PCollection。它需要指定 Kafka 主题、Kafka broker 地址、消费组 ID 等参数。示例代码如下：

PCollection> kafkaMessages = pipeline
  .apply(KafkaIO.read()
      .withBootstrapServers("broker_1:port,broker_2:port")
      .withKeyDeserializer(StringDeserializer.class)
      .withValueDeserializer(StringDeserializer.class)
      .withTopic("my_topic")
      .withConsumerConfigUpdates(
          ImmutableMap.of(
            "group.id", "my_beam_group"
          )
      )
  )
  .apply(Values.create())
  .apply(ParDo.of(new DoFn>() {
    @ProcessElement
    public void processElement(ProcessContext c) {
      c.output(KV.of(null, c.element()));
    }
  }));

KafkaConsume 是一个低级别的 Beam API，它提供了更细粒度的控制，并可更好地集成到已有的 Kafka 代码中。它需要自己编写 Kafka 消费者代码，并将其集成到 Beam 管道中。示例代码如下：

PCollection kafkaMessages = pipeline.apply(Create.of("dummy"))
  .apply(ParDo.of(new DoFn() {
    @ProcessElement
    public void processElement(ProcessContext c) {
      KafkaConsumer consumer = new KafkaConsumer<>(properties);
      consumer.subscribe(topic);

      while (true) {
        ConsumerRecords records = consumer.poll(Duration.ofMillis(500));
        for (ConsumerRecord record : records) {
          c.output(record.value());
        }
      }
    }
  }))
  .apply(Window.into(FixedWindows.of(Duration.standardSeconds(30))));

总的来说，如果你只需要简单地从 Kafka 中读取数据并将其转换为 PCollection，则可以使用 ReadFromKafka 并指定必要的参数。如果你需要更细粒度的控制，则可以使用 KafkaConsume 并自己编写