pyspark从hive导出数据到clickhouse

2023年1月13日09:55:22

需求

从hive导出数据到clickhouse 大概有两百个字段，两千多万数据，30G左右

实现

采用pyspark导入，先上代码

#_*_ coding=UTF-8 _*_
import os
import sys
from pyspark.sql import SparkSession


def sparksession_read_hive():  # .master("local[*]")
    spark = SparkSession.builder.appName('bi_pharaoh_mkt_lead_detail_to_clickhouse')\
        .enableHiveSupport().getOrCreate()
    hive_df = spark.sql("""select  
    place_type 
                ,sub_place_type 
                
    from  dm.test where pt = 20220314
    """)

    print(hive_df.show())
    print('sparkcontext_read_hive读取完毕，开始准备写入')
    properties = {'driver': 'ru.yandex.clickhouse.ClickHouseDriver',
                  "socket_timeout": "300000",
                  "rewriteBatchedStatements": "true",
                  "batchsize": "1000000",
                  "numPartitions": "8",
                  'user': 'user',
                  'password': 'pass',
                  'isolationLevel': 'NONE'
                  }

    hive_df.write.jdbc(url='jdbc:clickhouse://{url}:8123/analysis',
                       table='test_table', mode='append', properties=properties)
    print('sparksession_read_hive写入完成')
    spark.stop()


def spark_te():
    spark = SparkSession.builder.appName('clickhouse').getOrCreate()
    properties = {"driver": "ru.yandex.clickhouse.ClickHouseDriver",
                  "socket_timeout": "300000",
                  "rewriteBatchedStatements": "true",
                  "batchsize": "1000000",
                  "numPartitions": "8",
                  'user': 'user',
                  'password': 'pass'}
    df = spark.read.jdbc(url='jdbc:clickhouse://{url}:8123/analysis',
                         table='bi_pharaoh_mkt_lead_detail_da', properties=properties)
    # spark.read.jdbc(properties=properties)
    # print(spark.sql('select * from default.test'))
    print(df.show(10))


if __name__ == '__main__':
    sparksession_read_hive()
    # spark_te()

1 PySpark简介

Apache Spark是用Scala编程语言编写的。为了用Spark支持Python，Apache Spark社区发布了一个工具PySpark。使用PySpark，您也可以使用Python编程语言处理RDD。正是由于一个名为Py4j的库，他们才能实现这一目标。
这里不介绍PySpark的环境设置，主要介绍一些实例，以便快速上手。

2 PySpark - SparkContext

SparkContext是任何spark功能的入口点。当我们运行任何Spark应用程序时，会启动一个驱动程序，它具有main函数，并且此处启动了SparkContext。然后，驱动程序在工作节点上的执行程序内运行操作。

SparkContext使用Py4J启动JVM并创建JavaSparkContext。默认情况下，PySpark将SparkContext作为’sc’提供，因此创建新的SparkContext将不起作用。

3 PySpark - Sparksession

Sparksession是更上层的接口，是对sparkcontext和Spark的封装，目前官方推荐使用SparkSession作为入口，实际上SparkSession中包含了SparkContext中的大部分方法。

RDD和DataFrame是Spark中最常用的两种数据集合。Column和Row分别是DataFrame类型中的一列和一行。

RDD，弹性分布式数据集，集合内对象分布在不同的节点之上，可以并行执行操作，是spark应用程序的基本数据结构。RDD中的元素类型是非结构化的，可以是任意JVM对象。

DataFrame，类似于关系型数据库中的表，相比于RDD，其内部的每一条记录都是结构化的。一般而言，推荐优先使用这种数据结构，可以轻易的在其上使用sql类操作，与hive结合较好。

再详细解释下，上面sparksession_read_hive() 这个方法是写入的方法，spark_te()这个方法是读取clickhouse的方法。spark简单语法如下：
SparkSession是新版本推荐的入口api，所有的spark操作都需要SparkSession来执行，首先，创建一个入口类，(类似于python的实例化)
语法是这样的：SparkSession.builder 然后，有下面几个函数：

.master()：设置运行模式，即：本地模式还是yarn模式
appName()：顾名思义，设置名字
enableHiveSupport() ：这是hive接口函数，如果想要查询hive的表就需要执行这个函数
getOrCreate(：这是最终的函数，创建或者获取。

然后如果是同一个数据库拿这个实例直接执行sql就可以了。如果是不同数据库那就需要像我这么写了，spark.write().jdbc()
spark目前好像是与hive和mysql接口集成，即如果读写hive和mysql是不需要jdbc的方式的，其他的都需要。jdbc有四个参数，url table。mode(‘append’,‘overwrite’,‘error’,等) ，properties (连接的配置，以字典形式传入，其中‘driver’项是驱动，是固定的，连value一起都是固定传入，其他参数可以酌情使用),
spark还有其他函数，这里就不一一介绍了。

再说说踩过的坑，一开始配置好环境，环境配置可以参考
在pycharm中安装spark环境
 Hadoop安装教程 Mac版
这两个结合着看就行

坑点1:

报错 pyspark从hive导出数据到clickhouse
py4j.protocol.Py4JJavaError: An error occurred while calling o75.jdbc.
java.lang.ClassNotFoundException:ru.yandex.clickhouse.ClickHouseDriver
没有clickhouse的驱动，去下载一个驱动，放进spark下的jars中解决。

坑点2:可以读取，不可写入

这里耽误了好久，因为觉得已经能够读取了，那就肯定不是驱动的事，一直在找语法上的错误，最后在网上查到缺少guava-28.0-jre.jar包
错误信息为：
java.lang.NoClassDefFoundError: com/google/common/escape/Escapers
or
NoClassDefFoundError: Could not initialize class ru.yandex.clickhouse.ClickHouseUtil
去下载一个放进 spark中的jar下，解决，但是要注意，3.1spark自带一个guava-14.0.jar,但自带的这个jar版本不够，必须放进去更高的版本。测试几次好像最低要guava-16.jar才行。另外，网上有人遇到放进spark的jars路径下无法解决，可以试试其他路径，参考链接：
Caused by: java.lang.NoClassDefFoundError: com/google/common/escape/Escapers

坑点3:不能overwrite

这个坑目前未解决，只要我用overwrite模式，就会报错，错误信息很粗略，改用append模式就能够成功写入，现在采取的方案是在执行spark程序之前先用clickhouse_driver 进行清空表的操作，然后使用append的模式写入。另外说一下，好像好多etl工具的overwrite模式好像都不怎么靠谱，动不动就报错。

最后，成功写入

补充：在pycharm中添加spark的环境变量：
和上面文档中提示添加的路径变量没有任何关系，重新梳理一下pycarm中的配置步骤。

1.

创建项目，指定python解释器的版本

2.

配置项目依赖，
pyspark从hive导出数据到clickhouse
py4j 将Python代码转换为Java代码的库
pyspark Python的Spark编程依赖库

选择Add Content Root，在弹出的文件选择框中，选择Spark安装目录中的python文件夹下的lib目录中的py4j和pyspark的依赖文件，点击OK，将这两个依赖加入到当前项目的依赖库中。

3.

点击右上角的程序配置，配置环境变量，添加SPARK_HOME值为SPARK安装包的解压路径，PYSAPRK_PYTHON指定PySpark的Python命令，注意python版本为3.6+。 pyspark从hive导出数据到clickhouse

调用

spark-submit --master yarn --deploy-mode cluster --name dm_bi_pharaoh_mkt_lead_detail_a_test_task --queue analysis --conf spark.pyspark.python=python3 --driver-class-path /usr/lib/spark-current/driver-jars/ --num-executors 8 --executor-memory 4g --executor-cores 4 --driver-memory 4g --conf spark.sql.execution.arrow.enabled=true --conf spark.kryoserializer.buffer.max=2008 --conf spark.driver.extraClassPath=/usr/lib/spark-current/driver-jars/clickhouse-jdbc-0.2.4.jar --conf spark.executor.extraClassPath=/usr/lib/spark-current/driver-jars/clickhouse-jdbc-0.2.4.jar   /mnt/disk1/server/code/data-integration/script/huike/etl/core/dm/bi/bi_pharaoh_mkt_lead_detail_to_clickhouse.py

后续补充1

问题：要使用自定义模块，报错找不到模块
解决：在调用命令里添加 --py-files 参数，把自定义模块的文件（含路径）写在后面，如果有多个用英文逗号隔开。
问题：要读取配置文件，报错没有section
解决：实际错误是找不到配置文件，没有section=空文件=没有文件。在调用命令里添加 --files 参数把文件写在后面，如果有多个用英文逗号隔开。

注意：–py-files 文件后填写的模块在导入时不需要考虑路径的问题，直接from xxx import Xxxx ，在–files后写的文件使用时也不需要考虑路径，直接使用文件名即可

参考文档
将数据通过spark从hive导入到Clickhouse

需求

实现