如何并行运行多个Spark作业？单Spark作业对应单Oracle查询需同时执行

Great question! Running multiple Spark jobs in parallel—each tied to its own Oracle query—is a common need when you want to maximize throughput and leverage both your Spark cluster and Oracle DB efficiently. Let’s walk through the most reliable approaches, along with key considerations to avoid pitfalls:

Spark clusters (whether Standalone, YARN, or Kubernetes) are built to handle multiple concurrent jobs out of the box, as long as there’s available resources. The simplest way is to submit each job as a separate spark-submit process.

For example, using shell scripts to submit jobs in the background:

# Submit job 1 in the background
spark-submit --class com.yourcompany.Job1 --master yarn --deploy-mode cluster job1.jar &
# Submit job 2 in the background
spark-submit --class com.yourcompany.Job2 --master yarn --deploy-mode cluster job2.jar &
# Add more jobs as needed
wait

This lets the cluster scheduler (like YARN’s ResourceManager) handle distributing resources across jobs. Just make sure you don’t submit more jobs than your cluster can support—keep an eye on executor cores, memory, and queue limits.

If you prefer managing all jobs from one entry point, you can use a thread pool to spawn multiple threads, each creating its own isolated SparkSession to run an Oracle query job.

Here’s a Python example using concurrent.futures:

from concurrent.futures import ThreadPoolExecutor
from pyspark.sql import SparkSession

def run_oracle_spark_job(query, job_id):
    # Create a dedicated SparkSession for this job
    spark = SparkSession.builder \
        .appName(f"OracleQueryJob-{job_id}") \
        .config("spark.executor.cores", "2") \
        .config("spark.executor.memory", "4g") \
        .getOrCreate()
    
    try:
        # Load data from Oracle
        df = spark.read \
            .format("jdbc") \
            .option("url", "jdbc:oracle:thin:@//oracle-host:1521/DB_NAME") \
            .option("dbtable", f"({query}) AS temp_table") \
            .option("user", "DB_USER") \
            .option("password", "DB_PASS") \
            .load()
        
        # Add your processing/write logic here
        df.write.mode("overwrite").parquet(f"/output/path/job_{job_id}")
    finally:
        # Always stop the SparkSession to free resources
        spark.stop()

if __name__ == "__main__":
    # List of your Oracle queries and unique job IDs
    oracle_jobs = [
        ("SELECT * FROM sales WHERE date >= '2024-01-01'", "sales_jan"),
        ("SELECT * FROM inventory WHERE stock < 10", "low_stock"),
        ("SELECT customer_id, COUNT(*) FROM orders GROUP BY customer_id", "customer_order_count")
    ]
    
    # Adjust max_workers based on your cluster's capacity
    with ThreadPoolExecutor(max_workers=3) as executor:
        for query, job_id in oracle_jobs:
            executor.submit(run_oracle_spark_job, query, job_id)

Pro tip: Don’t set max_workers higher than the number of available executor slots in your cluster—you’ll just end up with jobs waiting in the queue.

For production environments where you need monitoring, error handling, and retries, use a workflow scheduler like Apache Airflow or Apache Livy.

• Apache Airflow: Define each Spark job as a SparkSubmitOperator in a DAG, then set the DAG’s concurrency to allow parallel execution. Example snippet:

  from airflow import DAG
  from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
  from datetime import datetime
  
  default_args = {
      'owner': 'data_team',
      'start_date': datetime(2024, 1, 1)
  }
  
  with DAG('parallel_oracle_spark_jobs', default_args=default_args, schedule_interval='@daily', concurrency=3) as dag:
      job1 = SparkSubmitOperator(
          task_id='sales_query_job',
          application='/path/to/job1.jar',
          conn_id='spark_default'
      )
  
      job2 = SparkSubmitOperator(
          task_id='inventory_query_job',
          application='/path/to/job2.jar',
          conn_id='spark_default'
      )
  
      job3 = SparkSubmitOperator(
          task_id='customer_query_job',
          application='/path/to/job3.jar',
          conn_id='spark_default'
      )
  
      # All jobs run in parallel
      [job1, job2, job3]
  

• Apache Livy: Submit multiple batch jobs to the Livy server via its REST API. Livy manages Spark sessions and ensures jobs run in parallel as resources allow.

• Oracle DB Load: Parallel queries can strain your Oracle database. Work with your DBA to set concurrent connection limits or use Oracle’s resource manager to prioritize queries.
• Resource Isolation: Assign dedicated cores/memory to each Spark job to avoid contention. Use cluster-specific tools (like YARN queues) to partition resources for different job groups.
• Error Handling: A single failing job shouldn’t break the entire workflow. Add try/except blocks in thread-based code, or use scheduler features like retries and alerting.
• Session Cleanup: Always stop SparkSessions after job completion to prevent memory leaks and resource exhaustion.

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