能否使用Apache Beam直接读取Json文件并存储为PCollection<Json>？

Absolutely! You can totally read JSON files into a PCollection<YourJsonModel> efficiently without dealing with unnecessary data transfers or duplication. Since you’re already using Apache Beam’s TextIO and PCollection, let’s walk through the best way to do this:

1. Define your JSON model class first #

First, create a plain old Java object (POJO) that matches the structure of your JSON data. Use a library like Jackson or Gson to handle serialization/deserialization—these are optimized and way less error-prone than writing custom parsers.

Example with Jackson:

import com.fasterxml.jackson.annotation.JsonProperty;

public class User {
    @JsonProperty("user_id")
    private String userId;
    @JsonProperty("full_name")
    private String fullName;
    @JsonProperty("email")
    private String email;

    // Required no-arg constructor for Jackson deserialization
    public User() {}

    // Getters and setters
    public String getUserId() { return userId; }
    public void setUserId(String userId) { this.userId = userId; }
    public String getFullName() { return fullName; }
    public void setFullName(String fullName) { this.fullName = fullName; }
    public String getEmail() { return email; }
    public void setEmail(String email) { this.email = email; }
}

2. Read + Parse in a single pipeline step (no extra transfers) #

The key to avoiding inefficient data transfers is to chain the text read and JSON parsing directly in the same pipeline. This way, the raw JSON strings are only fetched once from storage, then parsed in-memory into your model class—no intermediate writes to storage, no redundant data movement.

Here’s how to implement it with TextIO and MapElements:

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.SimpleFunction;
import com.fasterxml.jackson.databind.ObjectMapper;

public class JsonPipelineExample {
    // Reuse a single ObjectMapper instance for efficiency
    private static final ObjectMapper JSON_MAPPER = new ObjectMapper();

    public static void main(String[] args) {
        Pipeline pipeline = Pipeline.create();

        pipeline
            // Step 1: Read raw JSON lines from storage (only one data fetch)
            .apply("Read JSON Lines", TextIO.read().from("gs://your-bucket/path/*.json"))
            // Step 2: Parse each line directly into your model class (in-memory, no extra transfers)
            .apply("Parse JSON to User", MapElements.via(new SimpleFunction<String, User>() {
                @Override
                public User apply(String jsonLine) {
                    try {
                        return JSON_MAPPER.readValue(jsonLine, User.class);
                    } catch (Exception e) {
                        // Handle bad lines (log, skip, or throw—adjust based on your needs)
                        System.err.println("Skipping invalid JSON line: " + jsonLine);
                        return null; // Or use a Filter transform later to remove nulls
                    }
                }
            }))
            // Now you have a PCollection<User> ready for further processing
            .apply("Process Users", ...);

        pipeline.run().waitUntilFinish();
    }
}

3. Handle multi-line JSON or JSON arrays #

If your JSON files aren’t line-delimited (e.g., a single file with a large JSON array), use TextIO.readAll() to read entire files as single strings, then parse the array in one go. This still avoids extra transfers—each file is read once, then parsed in-memory:

import org.apache.beam.sdk.transforms.Flatten;
import com.fasterxml.jackson.core.type.TypeReference;
import java.util.List;

// ...

pipeline
    .apply("Read Entire JSON Files", TextIO.readAll().from("gs://your-bucket/path/*.json"))
    .apply("Parse JSON Array to Users", MapElements.via(new SimpleFunction<String, Iterable<User>>() {
        @Override
        public Iterable<User> apply(String jsonContent) {
            try {
                return JSON_MAPPER.readValue(jsonContent, new TypeReference<List<User>>() {});
            } catch (Exception e) {
                throw new RuntimeException("Failed to parse JSON file", e);
            }
        }
    }))
    .apply("Flatten User List", Flatten.iterables());

Critical Efficiency Tips #

• Never write intermediate data: Don’t save the raw JSON strings to storage and then re-read them for parsing—this is the main cause of inefficient transfers. Keep everything in the same pipeline.
• Reuse deserialization instances: ObjectMapper (Jackson) and Gson instances are expensive to create—reuse a single instance across your transforms.
• Handle errors gracefully: Skip or log bad lines instead of failing the entire pipeline, unless you need strict validation.

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