Pyspark的dataframe写入hive表

 文章最前： 我是Octopus，这个名字来源于我的中文名--章鱼；我热爱编程、热爱算法、热爱开源。所有源码在我的个人github ；这博客是记录我学习的点点滴滴，如果您对 Python、Java、AI、算法有兴趣，可以关注我的动态，一起学习，共同进步。

 相关文章：

1. 协同过滤推荐算法和基于内容推荐算法的区别？
2. 推荐算法的分类
3. MMOE原理介绍

文章目录：

1. 创建pyspark的DataFrame

2.hive表的结构

3.pyspark的DSL风格写入Hive表

1. 创建pyspark的DataFrame

import pandas as pd
temp_dict ={
    'id':[1,2,3,4,9],
    'date':['2022-05-01','2022-05-02','2022-05-03','2022-05-04','2022-05-05'],
    'vol':[68,45,899,45,32],
    'need':[-4,18,66,18,65],
    'need2':[-4,19,66,18,65]
}
tempdf = pd.DataFrame(temp_dict)
df = spark.createDataFrame(tempdf)

2.hive表的结构

CREATE TABLE if NOT EXISTS tmp.table_5_15 (
  id int comment 'id',
  date string comment 'date',
  vol INT comment 'vol',
  need int comment 'need',
  nee2 int comment 'need2'
)
partitioned by
(dt string COMMENT '分区字段')stored as parquettblproperties('parquet.compression'='SNAPPY'

3.pyspark的DSL风格写入Hive表

spark.sql("set hive.exec.dynamic.partition.mode = nonstrict")
spark.sql("set hive.exec.dynamic.partition=true")
df.write.format("Hive") \
        .mode('overwrite') \
        .partitionBy("dt") \
        .option("header", "false") \
        .option("delimiter", "\t") \
        .saveAsTable("tmp.table_5_15")

pySpark直接存储hive，这里的"dt"是分区字段
mode分为"overwrite"'和”append"
"append”是向表中添加数据
"overwrite"是重新建表再写，意味着会删除原本的所有数据，而不仅仅只删除当前分区的数据

4.DSL风格写入hive表不删除 其他分区

configs = [
    ('spark.app.name', 'algo2_spark2_demo'),
    ('spark.driver.memory', '4g'),
    ('spark.executor.memory', '4g'),
    ('spark.executor.instances', '2'),
    ('spark.executor.cores', '2'),
    ('spark.kryoserializer.buffer.max','128m'),
    ("hive.exec.dynamic.partition.mode", "nonstrict") ,
    ("spark.sql.sources.partitionOverwriteMode", "dynamic")
]

conf.setAll(configs)
sc = SparkContext.getOrCreate(conf=conf)
spark = HiveContext(sc)
df.write.insertInto("tmp.table_5_15", overwrite=True)

这个方式插入数据不会影响其他分区。