如何通过PySpark加载HDFS中存储的SQLite数据库表？

问题本质 #

SQLite的JDBC驱动仅支持本地文件系统路径，无法直接解析HDFS的URL。你当前的JDBC URL写法会被驱动误解析为本地路径（比如/home/hadoop/coding/hdfs:），导致文件不存在的报错。

解决方案 #

方案1：将HDFS文件下载到本地临时目录后读取

这是最直接的方法，先通过Hadoop命令将HDFS上的SQLite文件拉取到本地临时路径，再用JDBC读取本地文件：

from pyspark.sql import SparkSession
import subprocess
import os

# 初始化Spark会话
spark = SparkSession.builder\
           .config('spark.jars.packages', 'org.xerial/sqlite-jdbc:3.45.3.0')\
           .getOrCreate()

# 定义路径
hdfs_file_path = "/user/hadoop/project/dataset/a97be29a-1c3f-4fe9-afce-d4da4a8667ab"
local_temp_db = "/tmp/temp_sqlite.db"

# 从HDFS下载文件到本地
subprocess.run(["hadoop", "fs", "-get", hdfs_file_path, local_temp_db], check=True)

# 读取本地SQLite文件
jdbc_url = f"jdbc:sqlite:{local_temp_db}"
table_name = "RealTimeData"
df = spark.read.format('jdbc') \
        .options(driver='org.sqlite.JDBC', dbtable=table_name, url=jdbc_url)\
        .option("customSchema", "date STRING")\
        .load()

df.show(5)

# 清理本地临时文件
os.remove(local_temp_db)

方案2：用Python sqlite3结合Spark RDD处理（适合小文件）

如果SQLite文件体积较小，可以直接读取HDFS上的二进制文件内容，在内存中加载并处理：

from pyspark.sql import SparkSession
import sqlite3
import io
from pyspark.sql.types import StructType, StructField, StringType

spark = SparkSession.builder.getOrCreate()

# 定义自定义Schema（和原需求一致）
custom_schema = StructType([
    StructField("date", StringType(), True),
    # 根据实际表结构补充其他字段
])

table_name = "RealTimeData"

# 读取HDFS上的SQLite二进制文件
db_rdd = spark.sparkContext.binaryFiles("hdfs://localhost:9000/user/hadoop/project/dataset/a97be29a-1c3f-4fe9-afce-d4da4a8667ab")

# 定义数据处理函数
def process_db_file(file_tuple):
    _, file_content = file_tuple
    # 将二进制内容转为内存文件
    db_io = io.BytesIO(file_content)
    conn = sqlite3.connect(":memory:")
    # 附加内存中的SQLite数据库
    conn.execute("ATTACH DATABASE ? AS target_db", (db_io,))
    # 查询目标表数据
    cursor = conn.execute(f"SELECT * FROM target_db.{table_name}")
    rows = cursor.fetchall()
    conn.close()
    return rows

# 转换为Spark DataFrame
result_rdd = db_rdd.flatMap(process_db_file)
df = spark.createDataFrame(result_rdd, schema=custom_schema)

df.show(5)

注意事项 #

• SQLite是单文件型数据库，不支持分布式读写。如果文件体积较大，建议先将SQLite中的数据导出为Parquet、CSV等支持分布式存储的格式，再用Spark处理。
• 方案1中的临时路径需确保执行Spark的节点有读写权限，集群环境下可使用节点共享的临时存储目录。

内容的提问来源于stack exchange，提问作者Oussama Jmâa