Python代码转PySpark代码：pd.read_sql语句的替代实现及基于PYODBC的EDL数据提取方案

Got it, let's break down how to replace that pd.read_sql call with PySpark code, especially for your enterprise data lake (EDL) Hive connection scenario.

First, a key point: PySpark is built for distributed data processing, so we'll want to use its native JDBC/ODBC integration instead of going through PyODBC → Pandas → Spark (which is inefficient for large datasets). Here are your two main options, starting with the recommended one:

Recommended Approach: Use Spark's Native JDBC Integration #

This method lets Spark directly connect to your EDL Hive instance, avoiding the overhead of loading data into a local Pandas DataFrame first.

Step 1: Initialize a SparkSession #

Start by creating a SparkSession, which is the entry point for all PySpark operations. If your environment is already configured for Hive integration, this will automatically pick up your Hive settings:

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("EDL Hive Data Extraction") \
    .config("spark.sql.catalogImplementation", "hive")  # Enables Hive integration
    .getOrCreate()

Step 2: Execute Your SQL Query #

If your SparkSession is connected to Hive (via the config above), you can run your SQL query directly with spark.sql()—this is the closest equivalent to pd.read_sql for PySpark:

# Your existing query (example provided)
query = f'''
WITH trans AS (
    SELECT a.employee_name, a.employee_id
    FROM EMP
)
-- Add the rest of your query logic here
SELECT * FROM trans
'''

# Execute query and get a Spark DataFrame
transaction_df = spark.sql(query)

# Verify the result (equivalent to Pandas' head())
transaction_df.show(5)

If you need to use an ODBC DSN specifically (like your original DSN=Hive setup), use Spark's JDBC format with the ODBC URL:

jdbc_url = "jdbc:odbc:Hive"  # Matches your DSN name
connection_props = {
    "driver": "org.apache.hive.jdbc.HiveDriver",  # Use your actual ODBC driver class
    "autocommit": "true"
}

# Wrap your query in a subquery (required for Spark JDBC's dbtable parameter)
transaction_df = spark.read \
    .format("jdbc") \
    .option("url", jdbc_url) \
    .option("dbtable", f"({query}) AS temp_trans") \
    .options(**connection_props) \
    .load()

Alternative: PyODBC → Pandas → Spark (Not Recommended for Large Data) #

If you have to use PyODBC for some reason (e.g., specific access restrictions), you can read the data into a Pandas DataFrame first, then convert it to a Spark DataFrame. Note: This only works for small datasets, as Pandas runs in local memory.

import pyodbc
import pandas as pd
from pyspark.sql import SparkSession

# Initialize SparkSession
spark = SparkSession.builder.appName("EDL via PyODBC").getOrCreate()

# PyODBC connection (same as your original code)
connect_to_hive = pyodbc.connect("DSN=Hive", autocommit=True)
# Read to Pandas DataFrame
transaction_pd = pd.read_sql(query, connect_to_hive)
# Convert to Spark DataFrame
transaction_df = spark.createDataFrame(transaction_pd)
# Close connection
connect_to_hive.close()

Key Notes to Keep in Mind #

• Driver Setup: For Spark's JDBC/ODBC integration, ensure your Spark cluster has the required Hive JDBC/ODBC drivers installed (either in Spark's classpath or passed via --jars when submitting jobs).
• Data Size: Always prefer the native Spark JDBC approach for large datasets—Pandas can't handle data that exceeds your local machine's memory.
• API Differences: Spark DataFrames use PySpark's API (e.g., transaction_df.write.parquet() instead of transaction_pd.to_parquet()), so you'll need to adjust any downstream processing code.

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