pyspark连接clickhouse数据库的方式(其它数据库同样适用）

news2025/1/21 19:24:49

一、背景简记

实际工作中，大部分所用的数据存储地址都是在数据库中，如我司现在常用的数据库是clickhouse，相关数据的统计分析都在此上操作。如果想用pyspark连接clickhouse数据库怎么办呢，经过博主的一番探索，由此衍生了本文。
当然，本文连接clickhouse数据库的方式，同样可以衍生使用与mysql、Oracle，postgre数据库等。

在这里插入图片描述

二、pyspark连接clickhouse方式记录

如果按照上一篇博文《Windows环境本地配置pyspark环境详细教程》，安装配置好了本地pyspark环境，则本地连接clickhouse数据库的脚本示例如下。

文件 《clickhouse-jdbc-0.6.1.jar》 是本地连接clickhouse数据库的驱动，这在通过DBeaver或者Navicat连接clickhouse数据库时，即会下载该驱动，因此使用使用搜索确认下该文件地址路径即可。
脚本中相关注释掉的代码，如果本地未设定jdk、spark环境变量，可以在脚本中内置指定其路径。
标*的为数据库的ip地址和连接的数据库名称，实际根据情况填写即可。

# import findspark
# findspark.init() # 初始化本地环境
 import os
from pyspark.sql import SparkSession
# 设置环境变量
# os.environ["JAVA_HOME"] = "D:/JDK" # 手动指定jdk安装位置
# os.environ["SPARK_HOME"] = "D:/bigdataenv/spark-3.5.0-bin-hadoop3" # 手动指定spark安装位置
# os.environ['PYSPARK_SUBMIT_ARGS'] = '--master local[*] --conf spark.driver.memory=4g --conf spark.executor.memory=2g --conf spark.driver.extraJavaOptions=-Djava.library.path="D:/JDK/bin" pyspark-shell' # 相关提交信息设置，注意其中含有jdk位置

# 创建 SparkSession
spark = SparkSession.builder \
    .appName("ClickHouse Connection") \
    .config("spark.jars", "file:///C:/Users/Administrator/AppData/Roaming/DBeaverData/drivers/maven/maven-central/com.clickhouse/clickhouse-jdbc-0.6.1.jar") \
    .getOrCreate()

# 获取 SparkContext
sc = spark.sparkContext

# 设置日志级别
sc.setLogLevel("ERROR")

# # 测试 SparkSession 是否正常工作
# df = spark.createDataFrame([(1, 'foo'), (2, 'bar')], ['id', 'value'])

# ClickHouse连接参数
clickhouse_url = "jdbc:clickhouse://**.**.**.**:8123/sj**"
properties = {
    "user": "default",
    "password": "123456",
    "driver": "com.clickhouse.jdbc.ClickHouseDriver"
}

# 读取数据
df = spark.read.jdbc(url=clickhouse_url, table="ods_bt_base_dict", properties=properties)

df.show() 
spark.stop()

结果如下图所示：
在这里插入图片描述

记录一个本地测试的示例：

import findspark
findspark.init() 
# findspark.init(spark_home="D:\\bigdataenv\\\spark-3.5.0-bin-hadoop3") #  # 自动查找 SPARK_HOME

from pyspark.sql import SparkSession

# 创建 Spark 会话
spark = SparkSession.builder \
    .appName("Test PySpark") \
    .master("local[*]") \
    .getOrCreate()

# 打印 Spark 会话信息
print("Spark version:", spark.version)

# 创建一个 DataFrame
data = [("Alice", 34), ("Bob", 45), ("Cathy", 29)]
columns = ["Name", "Age"]
df = spark.createDataFrame(data, columns)

# 显示 DataFrame
df.show()

# 停止 Spark 会话
spark.stop()