连接mysql查询数据:

初始化mysql的连接信息

def init_mysql():
url = “jdbc:mysql://ip:port/databasename?&useSSL=false&serverTimezone=GMT%2B8”
properties = {“user”: “user”
, “password”: “password”
, “driver”: “com.mysql.cj.jdbc.Driver”
}
return url, properties

使用mysql的连接信息查询mysql数据

sql = “(select * from table) table”
def get_table_data(spark, url, sql, temp_table, properties):
df = spark.read.jdbc(url=url, table=sql, properties=properties)
df.createOrReplaceTempView(temp_table)
return df

插入数据库的两种方式

save方式

(df.write.format(“jdbc”)
.option(“url”, “jdbc:mysql://ip:port/databasename”)
.option(“driver”, “com.mysql.cj.jdbc.Driver”)
.option(“user”, “user”)
.option(“password”, “password”)
.option(“dbtable”, “table”)
.mode(saveMode=‘append’)#overwrite会删除表不建议使用,append可以向表中追加数据
.save())

jdbc函数

df.write.jdbc(url, table, “append”, properties)