说明

这与 传统方式：连接2个数据源不同，这种是 先连接 Hive进行查询，返回 DataFrame 对象，通过 DataFrame.write.jdbc() 将查询结果保存到 MySQL 中

要求

将 mysql 驱动放入到spark的 jars 目录

语法

官网 API

DataFrameWriter.jdbc(url, table, mode=None, properties=None)

解释：

• url：数据库 url
• table：要保存数据的表
• mode：模式，有以下几个值：
  • append: 将数据保存到表中。MySQL中的表不存在，会自动创建
  • overwrite: 如果表存在，则 删除表，然后创建表，将数据保存到该表中
  • ignore: 表存在，则不做任何操作：不写入数据、也不报错
  • error or errorifexists (default case): 默认值，如果表存在，则报错
• properties：dict 类型，配置 用户名、密码、驱动

关键代码

prop = {"user": "root","password": "root","driver": "com.mysql.jdbc.Driver"}
mysqlURL = "jdbc:mysql://hadoop1:3306/yuqing_dws?useSSL=false&characterEncoding=utf8"
df.write.jdbc(mysqlURL, "t_gender_statistics","append", prop)

注意：需要设置编码集 characterEncoding=utf8，否则保存中文会报错

完整代码

import findspark
import datetime

findspark.init()

from pyspark.sql import  SparkSession 


def exe():
    spark = SparkSession.builder.\
    appName("test").\
    master("local[*]").\
    enableHiveSupport().\
    getOrCreate()
    # 查询 hive
    df = spark.sql("select gender,count(1) as `count`,'weibo' as oragin,substr(current_timestamp(),1,19) as `create_time` from dwd_yuqing.t_weibo group by gender")
    # df = spark.sql("select 'f' as gender,1 as `count`,'weibo' as oragin,substr(current_timestamp(),1,19) as `create_time`")

    df.show()
    df.printSchema()
    # 将 查询结果保存到 MySQL数据库中
    prop = {"user": "root","password": "root","driver": "com.mysql.jdbc.Driver"}
    mysqlURL = "jdbc:mysql://hadoop1:3306/yuqing_dws?useSSL=false&characterEncoding=utf8"
    df.write.jdbc(mysqlURL, "t_gender_statistics","append", prop)

    spark.stop()

if __name__ == "__main__":
    exe()

原文出处：/show_1IX3H4cMRafL.html