SparkSession与JavaStreamingContext冲突，求基于SparkSession的Kinesis结构化流示例

Hey there! I totally get where you're coming from—mixing the legacy JavaStreamingContext with SparkSession can lead to weird conflicts since they're built on different APIs. The good news is that Structured Streaming (powered by SparkSession) is the way to go for modern Spark applications, and it works seamlessly with Kinesis while making database interactions a breeze. Let's walk through a complete, working Java example tailored to your needs.

Step 1: Add Required Dependencies #

First, make sure your pom.xml (if using Maven) includes these dependencies. You'll need Spark SQL, the Kinesis connector, and any JDBC driver for your database (I'll use PostgreSQL as an example):

<dependencies>
    <!-- Spark SQL (includes SparkSession) -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.12</artifactId>
        <version>3.5.0</version>
        <scope>provided</scope>
    </dependency>
    <!-- Kinesis Connector for Structured Streaming -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql-kinesis_2.12</artifactId>
        <version>3.5.0</version>
    </dependency>
    <!-- JDBC Driver for PostgreSQL (adjust for your DB) -->
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>postgresql</artifactId>
        <version>42.6.0</version>
    </dependency>
</dependencies>

If you're using Gradle, translate those to the equivalent dependency declarations.

Step 2: Complete Java Code Example #

Here's a full example that reads from Kinesis, processes the data, and writes it to a database using SparkSession. I've added comments to explain each part clearly:

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;

import java.util.Properties;

public class KinesisStructuredStreamingExample {
    public static void main(String[] args) throws StreamingQueryException {
        // 1. Create SparkSession (the single entry point for all Structured Streaming operations)
        SparkSession spark = SparkSession.builder()
                .appName("KinesisToDatabaseExample")
                .master("local[*]") // Remove this line when running on a Spark cluster
                .getOrCreate();

        // 2. Configure Kinesis connection parameters
        String kinesisStreamName = "your-kinesis-stream-name";
        String kinesisEndpointUrl = "https://kinesis.us-east-1.amazonaws.com"; // Adjust to your AWS region
        String awsAccessKey = "your-aws-access-key";
        String awsSecretKey = "your-aws-secret-key";
        String startingPosition = "latest"; // Use "trim_horizon" to read from the beginning of the stream

        // 3. Read streaming data from Kinesis
        Dataset<Row> kinesisDF = spark.readStream()
                .format("kinesis")
                .option("streamName", kinesisStreamName)
                .option("endpointUrl", kinesisEndpointUrl)
                .option("awsAccessKey", awsAccessKey)
                .option("awsSecretKey", awsSecretKey)
                .option("startingPosition", startingPosition)
                .load();

        // 4. Process the raw Kinesis data: convert binary data to string first
        Dataset<Row> processedDF = kinesisDF
                .selectExpr("CAST(data AS STRING) as message")
                // Add your custom processing here (e.g., parse JSON, filter invalid records, transform fields)
                .filter("message IS NOT NULL");

        // 5. Configure database connection (PostgreSQL example)
        Properties dbProps = new Properties();
        dbProps.setProperty("user", "db-username");
        dbProps.setProperty("password", "db-password");
        String jdbcUrl = "jdbc:postgresql://your-db-host:5432/your-db-name";
        String targetTable = "your-target-table";

        // 6. Write streaming data to the database
        StreamingQuery query = processedDF.writeStream()
                .foreachBatch((batchDF, batchId) -> {
                    // Use Spark's DataFrame API to write batch data to JDBC
                    batchDF.write()
                            .mode("append")
                            .jdbc(jdbcUrl, targetTable, dbProps);
                })
                .outputMode("append")
                .start();

        // Keep the application running until manually terminated
        query.awaitTermination();
    }
}

Key Tips for Production & Troubleshooting #

• AWS Credentials: Never hardcode credentials in your code! Use environment variables, AWS IAM roles (if running on EC2/EKS), or the AWS credentials file in production.
• Cluster Execution: When deploying to a Spark cluster, remove the .master("local[*]") line and submit the app with the Kinesis connector package:

  spark-submit --class KinesisStructuredStreamingExample \
    --packages org.apache.spark:spark-sql-kinesis_2.12:3.5.0 \
    your-app-jar-file.jar
  

• JSON Data Handling: If your Kinesis records are JSON, use Spark's from_json function to parse the string into structured columns (e.g., select(from_json(col("message"), schema).alias("data"))).
• Output Modes: Adjust outputMode based on your use case—append works for inserting new records, update for updating existing ones, and complete for full table overwrites.

Why This Beats Legacy Streaming #

• Unified API: SparkSession lets you use the same DataFrame API for both streaming (Kinesis) and batch (database) operations—no more context switching or conflict issues.
• Fault Tolerance: Structured Streaming provides exactly-once semantics out of the box, so you don't have to worry about duplicate or lost data.
• Rich Features: You get access to all Spark SQL tools (window functions, joins, UDFs) directly in your streaming pipeline, making complex processing way easier.

I hope this example gets you up and running smoothly! If you hit any snags with specific parts (like parsing JSON or connecting to your database), feel free to tweak the code or ask for more details.

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