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

一、背景简记

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

在这里插入图片描述

clickhouse_10">二、pyspark连接clickhouse方式记录

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

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

python"># 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()

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

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

python">import findspark
findspark.init() 
# findspark.init(spark_home="D:\\bigdataenv\\\spark-3.5.0-bin-hadoop3") #  # 自动查找 SPARK_HOMEfrom 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()