Spark同时写入MongoDB与MySQL问题求助

Hey there! Let's work through this together—since you already have Spark writing to MySQL sorted, adding MongoDB into the mix is totally doable. The key is to split your pipeline to handle both the raw Twitter data (for MongoDB) and the flattened data (for MySQL) without stepping on each other's toes. Here's a step-by-step breakdown:

MongoDB needs its Spark Connector to play nice with your cluster. You'll need to include this in your spark-submit command—make sure the version matches your Spark and Scala setup (for example, Spark 3.3.x uses Scala 2.12 and connector version 3.0.x).

Your existing JSON flattening code is for MySQL, but we need to keep the raw Twitter data intact for MongoDB. Here's how to modify your convertMongo function to handle both:

import json

def convertMongo(rdd):
    try:
        spark = getSparkSessionInstance(rdd.context.getConf())
        
        # Cache the RDD to avoid reprocessing raw data twice (optional but efficient)
        rdd = rdd.cache()
        
        # --------------------------
        # Write raw Twitter data to MongoDB
        # --------------------------
        # Convert raw JSON strings into a DataFrame (MongoDB works great with nested docs)
        raw_tweet_df = spark.createDataFrame(
            rdd.map(lambda x: {"raw_tweet": json.loads(x[1])})  # Parse raw JSON into a nested structure
            # Or store the raw string directly if needed: {"raw_tweet": x[1]}
        )
        
        # Configure MongoDB connection and write
        raw_tweet_df.write \
            .format("mongo") \
            .option("uri", "mongodb://<your-mongo-host>:27017/<your-db>.<raw-tweets-collection>") \
            # Add auth options if required: .option("username", "user").option("password", "pass")
            .mode("append")  # Use "overwrite" only if you want to replace existing data
            .save()
        
        # --------------------------
        # Write flattened data to MySQL (your existing logic)
        # --------------------------
        flattened_rdd = rdd.map(lambda x: _flatten_JSON(json.loads(x[1])))
        flattened_df = spark.createDataFrame(flattened_rdd)
        
        # Your existing MySQL write code
        flattened_df.write \
            .format("jdbc") \
            .option("url", "jdbc:mysql://<your-mysql-host>:3306/<your-db>") \
            .option("dbtable", "<your-flattened-table>") \
            .option("user", "<mysql-user>") \
            .option("password", "<mysql-pass>") \
            .option("driver", "com.mysql.cj.jdbc.Driver")  # Use the modern MySQL driver
            .mode("append") \
            .save()
            
    except Exception as e:
        print(f"Pipeline failed with error: {str(e)}")
        raise e
    finally:
        # Uncache the RDD to free up cluster memory
        if rdd.is_cached:
            rdd.unpersist()

Include both the MongoDB and MySQL connectors as packages to avoid dependency conflicts:

spark-submit \
  --packages org.mongodb.spark:mongo-spark-connector_2.12:3.0.1,mysql:mysql-connector-java:8.0.33 \
  --class com.your.package.YourMainClass \
  your-application-jar.jar

Quick Tips to Avoid Headaches: #

• Version Alignment: Double-check that the MongoDB connector version matches your Spark version (e.g., Spark 3.2 uses connector 2.4.x).
• Error Isolation: Wrap each write operation in its own try-except block if you don't want a failed MongoDB write to take down the entire MySQL pipeline.
• MongoDB Flexibility: Since MongoDB supports nested JSON, you don't need to flatten raw tweets—store them as-is to preserve all original metadata from the Twitter data.

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