Spark Dataset API技术咨询：多行JSON读取与Dataset连接条件设置

Hey there! Let's break down your two Spark questions with Java code examples that align with your existing snippet.

1. Reading Multi-Line JSON Data with Spark Dataset API #

Right now, you're using read().text() which reads raw text lines, but Spark has a built-in way to handle multi-line JSON directly. The default JSON reader expects one JSON object per line—so to parse JSON that spans multiple lines (like a single object with nested structures split across lines), you need to enable the multiLine option.

Here's how to adjust your code:

public static void main(String[] args) {
    SparkSession spark = SparkSession
        .builder()
        .appName("Java Spark SQL basic example")
        .master("spark://10.127.153.198:7077")
        .getOrCreate();

    // Read multi-line JSON into a Dataset<Row>
    Dataset<Row> df = spark.read()
        .option("multiLine", true) // Enable multi-line parsing
        .json("C:\\Users\\phyadavi\\LearningAndDevelopment\\Spark-Demo\\your-multi-line-file.json");

    // Optional: Verify the result
    df.printSchema();
    df.show();
}

• The multiLine flag tells Spark to treat the entire file as a single JSON object (or array of objects) instead of parsing line-by-line.
• If your file contains multiple multi-line JSON objects back-to-back, add option("mode", "PERMISSIVE") to handle parsing errors gracefully.

2. Joining Two Datasets with join(): Specifying Conditions/Columns #

Spark's join() method has several overloads to handle different join scenarios. Let's cover the most common cases using Java:

Case 1: Join on a Single Matching Column Name #

If both Datasets share a column with the same name (e.g., partyId), you can pass the column name directly:

// Assume df1 and df2 both have a "partyId" column
Dataset<Row> joinedDf = df1.join(df2, "partyId");
// This defaults to an inner join

Case 2: Join on Multiple Matching Columns #

For multiple shared columns, pass a list of column names:

List<String> joinColumns = Arrays.asList("partyId", "eventDate");
Dataset<Row> joinedDf = df1.join(df2, joinColumns);

Case 3: Join on Custom Conditions (Different Column Names or Complex Logic) #

If columns have different names (e.g., df1 has partyId, df2 has party_id) or you need complex logic, use a Column expression:

// Join on differently named columns
Dataset<Row> joinedDf = df1.join(df2, df1.col("partyId").equalTo(df2.col("party_id")));

// Specify a join type (e.g., left outer join)
Dataset<Row> leftJoinedDf = df1.join(df2, df1.col("partyId").equalTo(df2.col("party_id")), "left_outer");

Supported join types include: inner, outer, full, full_outer, left, left_outer, right, right_outer, left_semi, left_anti.

Full Example with Your Code Context #

Here's how it might fit into your existing main method:

public static void main(String[] args) {
    SparkSession spark = SparkSession
        .builder()
        .appName("Java Spark SQL basic example")
        .master("spark://10.127.153.198:7077")
        .getOrCreate();

    // Read first multi-line JSON dataset
    Dataset<Row> df1 = spark.read()
        .option("multiLine", true)
        .json("path/to/first-dataset.json");

    // Read second dataset
    Dataset<Row> df2 = spark.read()
        .option("multiLine", true)
        .json("path/to/second-dataset.json");

    // Join on partyId match (inner join)
    Dataset<Row> joinedDf = df1.join(df2, df1.col("partyId").equalTo(df2.col("partyId")), "inner");

    joinedDf.show();
}

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