有Apache Kafka经验，求Java实现Apache Beam流式处理JSON数据示例指导

Hey there! Let's break this down for you step by step—leaning into your existing Kafka experience to make the Beam comparison clear, then walking through a Java implementation for your JSON data.

Since you’ve worked with Kafka Producers/Consumers, let’s ground this in what you already know:

• Core Purpose:
  • Kafka: A distributed streaming platform focused on message storage and real-time transport. It’s the backbone for moving data between systems—think of it as a high-throughput, persistent data bus. You use Producers to push data into topics, Consumers to pull it out, which is perfect for pub/sub or point-to-point data flows.
  • Apache Beam: A unified programming model for building batch and streaming data processing pipelines. It doesn’t handle data storage/transport itself; instead, it works with runners like Kafka Streams, Flink, or Dataflow to execute your processing logic. The big win here is writing once, running anywhere for complex transformations, aggregations, or windowing.
• Processing Capabilities:
  • Kafka: Basic stateful processing via Kafka Streams (tied closely to Kafka’s ecosystem). Great for simple tasks like filtering, mapping, or joining streams from Kafka topics.
  • Beam: Supports advanced processing patterns across both batch and streaming—event-time windowing (tumbling, sliding, session), side inputs, cross-source joins, and portability across runners. You can mix data from Kafka, files, or cloud pub/sub without rewriting core logic.
• Use Cases:
  • Kafka: Ideal for reliable data ingestion, microservices message queues, or as a central data bus for your system.
  • Beam: Best for end-to-end pipelines where you need to combine multiple data sources, apply complex transformations, and run the same pipeline for historical batch backfills and real-time streaming.

Let’s build a pipeline that processes your sample JSON. We’ll use Jackson for parsing and Beam’s DirectRunner (great for local testing).

Step 1: Add Dependencies #

Include these in your pom.xml (Maven):

<dependencies>
    <!-- Apache Beam Core -->
    <dependency>
        <groupId>org.apache.beam</groupId>
        <artifactId>beam-sdks-java-core</artifactId>
        <version>2.54.0</version> <!-- Use latest stable version -->
    </dependency>
    <!-- Beam Direct Runner for local testing -->
    <dependency>
        <groupId>org.apache.beam</groupId>
        <artifactId>beam-runners-direct-java</artifactId>
        <version>2.54.0</version>
        <scope>runtime</scope>
    </dependency>
    <!-- Jackson for JSON parsing -->
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
        <version>2.15.2</version>
    </dependency>
</dependencies>

Step 2: Define a POJO for JSON Mapping #

Create a class to map your JSON structure:

import com.fasterxml.jackson.annotation.JsonProperty;

public class UserData {
    @JsonProperty("UserID")
    private String userId;
    @JsonProperty("Address")
    private String address;
    @JsonProperty("ClassNo")
    private String classNo;
    @JsonProperty("UserName")
    private String userName;
    @JsonProperty("ClassType")
    private String classType;

    // Getters and setters required for Jackson deserialization
    public String getUserId() { return userId; }
    public void setUserId(String userId) { this.userId = userId; }
    public String getAddress() { return address; }
    public void setAddress(String address) { this.address = address; }
    public String getClassNo() { return classNo; }
    public void setClassNo(String classNo) { this.classNo = classNo; }
    public String getUserName() { return userName; }
    public void setUserName(String userName) { this.userName = userName; }
    public String getClassType() { return classType; }
    public void setClassType(String classType) { this.classType = classType; }

    // Optional: Make output readable
    @Override
    public String toString() {
        return "UserData{" +
                "userId='" + userId + '\'' +
                ", userName='" + userName + '\'' +
                ", classNo='" + classNo + '\'' +
                '}';
    }
}

Step 3: Build the Complete Pipeline #

This pipeline generates your sample JSON, parses it, applies a simple filter, and prints results:

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import com.fasterxml.jackson.databind.ObjectMapper;

public class JsonStreamingPipeline {
    public static void main(String[] args) {
        // Initialize pipeline for local testing
        PipelineOptions options = PipelineOptionsFactory.create();
        Pipeline pipeline = Pipeline.create(options);

        // Your sample JSON data (replace with Kafka input later)
        String sampleJson = "{\"UserID\":\"1\",\"Address\":\"XXX\",\"ClassNo\":\"989\",\"UserName\":\"Stella\",\"ClassType\":\"YYY\"}";

        pipeline
                // 1. Create a stream of sample JSON data
                .apply("Create Sample Data", Create.of(sampleJson))
                // 2. Parse JSON to UserData POJO
                .apply("Parse JSON", ParDo.of(new DoFn<String, UserData>() {
                    private final ObjectMapper objectMapper = new ObjectMapper();

                    @ProcessElement
                    public void processElement(@Element String json, OutputReceiver<UserData> out) {
                        try {
                            UserData user = objectMapper.readValue(json, UserData.class);
                            out.output(user);
                        } catch (Exception e) {
                            System.err.println("Failed to parse JSON: " + json + " | Error: " + e.getMessage());
                        }
                    }
                }))
                // 3. Example transformation: Filter users in ClassNo 989
                .apply("Filter Class 989", ParDo.of(new DoFn<UserData, UserData>() {
                    @ProcessElement
                    public void processElement(@Element UserData user, OutputReceiver<UserData> out) {
                        if ("989".equals(user.getClassNo())) {
                            out.output(user);
                        }
                    }
                }))
                // 4. Convert back to JSON for output
                .apply("Convert to JSON", ParDo.of(new DoFn<UserData, String>() {
                    private final ObjectMapper objectMapper = new ObjectMapper();

                    @ProcessElement
                    public void processElement(@Element UserData user, OutputReceiver<String> out) {
                        try {
                            String json = objectMapper.writeValueAsString(user);
                            out.output(json);
                        } catch (Exception e) {
                            System.err.println("Failed to convert to JSON: " + user + " | Error: " + e.getMessage());
                        }
                    }
                }))
                // 5. Print results to console
                .apply("Print Output", ParDo.of(new DoFn<String, Void>() {
                    @ProcessElement
                    public void processElement(@Element String json) {
                        System.out.println("Processed Result: " + json);
                    }
                }));

        // Run the pipeline
        pipeline.run().waitUntilFinish();
    }
}

Integrating with Kafka #

To read from Kafka instead of sample data, replace the Create step with:

import org.apache.beam.sdk.io.kafka.KafkaIO;
import org.apache.kafka.common.serialization.StringDeserializer;

// ...

.apply("Read from Kafka", KafkaIO.<String, String>read()
        .withBootstrapServers("your-kafka-broker:9092")
        .withTopic("your-input-topic")
        .withKeyDeserializer(StringDeserializer.class)
        .withValueDeserializer(StringDeserializer.class)
        .withoutMetadata())
.apply("Extract JSON Value", ParDo.of(new DoFn<KafkaRecord<String, String>, String>() {
    @ProcessElement
    public void processElement(@Element KafkaRecord<String, String> record, OutputReceiver<String> out) {
        out.output(record.getValue());
    }
}))

To write processed data back to Kafka, replace the Print Output step with:

import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.beam.sdk.io.kafka.VoidSerializer;

// ...

.apply("Write to Kafka", KafkaIO.<Void, String>write()
        .withBootstrapServers("your-kafka-broker:9092")
        .withTopic("your-output-topic")
        .withKeySerializer(VoidSerializer.class)
        .withValueSerializer(StringSerializer.class))

内容的提问来源于stack exchange，提问作者Stella