快捷導(dǎo)航

在python中使用pyspark讀寫(xiě)Hive數(shù)據(jù)操作

更新時(shí)間：2020年06月06日 14:11:40 作者：_____miss

這篇文章主要介紹了在python中使用pyspark讀寫(xiě)Hive數(shù)據(jù)操作，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧

1、讀Hive表數(shù)據(jù)

pyspark讀取hive數(shù)據(jù)非常簡(jiǎn)單，因?yàn)樗袑?zhuān)門(mén)的接口來(lái)讀取，完全不需要像hbase那樣，需要做很多配置，pyspark提供的操作hive的接口，使得程序可以直接使用SQL語(yǔ)句從hive里面查詢(xún)需要的數(shù)據(jù)，代碼如下：

from pyspark.sql import HiveContext,SparkSession
 
_SPARK_HOST = "spark://spark-master:7077"
_APP_NAME = "test"
spark_session = SparkSession.builder.master(_SPARK_HOST).appName(_APP_NAME).getOrCreate()
 
hive_context= HiveContext(spark_session )
 
# 生成查詢(xún)的SQL語(yǔ)句，這個(gè)跟hive的查詢(xún)語(yǔ)句一樣，所以也可以加where等條件語(yǔ)句
hive_database = "database1"
hive_table = "test"
hive_read = "select * from {}.{}".format(hive_database, hive_table)
 
# 通過(guò)SQL語(yǔ)句在hive中查詢(xún)的數(shù)據(jù)直接是dataframe的形式
read_df = hive_context.sql(hive_read)

2 、將數(shù)據(jù)寫(xiě)入hive表

pyspark寫(xiě)hive表有兩種方式：

（1）通過(guò)SQL語(yǔ)句生成表

from pyspark.sql import SparkSession, HiveContext
 
_SPARK_HOST = "spark://spark-master:7077"
_APP_NAME = "test"
 
spark = SparkSession.builder.master(_SPARK_HOST).appName(_APP_NAME).getOrCreate()
 
data = [
 (1,"3","145"),
 (1,"4","146"),
 (1,"5","25"),
 (1,"6","26"),
 (2,"32","32"),
 (2,"8","134"),
 (2,"8","134"),
 (2,"9","137")
]
df = spark.createDataFrame(data, ['id', "test_id", 'camera_id'])
 
# method one，default是默認(rèn)數(shù)據(jù)庫(kù)的名字，write_test 是要寫(xiě)到default中數(shù)據(jù)表的名字
df.registerTempTable('test_hive')
sqlContext.sql("create table default.write_test select * from test_hive")

（2）saveastable的方式

# method two
 
# "overwrite"是重寫(xiě)表的模式，如果表存在，就覆蓋掉原始數(shù)據(jù)，如果不存在就重新生成一張表
# mode("append")是在原有表的基礎(chǔ)上進(jìn)行添加數(shù)據(jù)
df.write.format("hive").mode("overwrite").saveAsTable('default.write_test')

tips：

spark用上面幾種方式讀寫(xiě)hive時(shí)，需要在提交任務(wù)時(shí)加上相應(yīng)的配置,不然會(huì)報(bào)錯(cuò)：

spark-submit --conf spark.sql.catalogImplementation=hive test.py

補(bǔ)充知識(shí)：PySpark基于SHC框架讀取HBase數(shù)據(jù)并轉(zhuǎn)成DataFrame

一、首先需要將HBase目錄lib下的jar包以及SHC的jar包復(fù)制到所有節(jié)點(diǎn)的Spark目錄lib下

二、修改spark-defaults.conf 在spark.driver.extraClassPath和spark.executor.extraClassPath把上述jar包所在路徑加進(jìn)去

三、重啟集群

四、代碼

#/usr/bin/python
#-*- coding:utf-8 –*-
 
from pyspark import SparkContext
from pyspark.sql import SQLContext,HiveContext,SparkSession
from pyspark.sql.types import Row,StringType,StructField,StringType,IntegerType
from pyspark.sql.dataframe import DataFrame
 
sc = SparkContext(appName="pyspark_hbase")
sql_sc = SQLContext(sc)
 
dep = "org.apache.spark.sql.execution.datasources.hbase"
#定義schema
catalog = """{
       "table":{"namespace":"default", "name":"teacher"},
       "rowkey":"key",
       "columns":{
            "id":{"cf":"rowkey", "col":"key", "type":"string"},
            "name":{"cf":"teacherInfo", "col":"name", "type":"string"},
            "age":{"cf":"teacherInfo", "col":"age", "type":"string"},
            "gender":{"cf":"teacherInfo", "col":"gender","type":"string"},
            "cat":{"cf":"teacherInfo", "col":"cat","type":"string"},
            "tag":{"cf":"teacherInfo", "col":"tag", "type":"string"},
            "level":{"cf":"teacherInfo", "col":"level","type":"string"} }
      }"""
 
df = sql_sc.read.options(catalog = catalog).format(dep).load()
 
print ('***************************************************************')
print ('***************************************************************')
print ('***************************************************************')
df.show()
print ('***************************************************************')
print ('***************************************************************')
print ('***************************************************************')
sc.stop()

五、解釋

數(shù)據(jù)來(lái)源參考請(qǐng)本人之前的文章，在此不做贅述

schema定義參考如圖：