pyspark写法总结

红太狼 2022-06-13 22:56 178阅读 0赞

## 往hive表中插入数据 ##

法1：dataFrame数据写入hive表

def log2Hive():

log=hiveContext.createDataFrame(\[\{"dt":dt,"types":types,"message":msg,"currtime":currTime\}\]).coalesce(1)\#types: "INFO" ,"ERROR"

log.write.mode("append").insertInto("app.app\_tion\_log")

法2：dataFrame数据写入hive表

temp\_vl\_org=self.vl\_data.join(self.order\_slice,\["po\_no"\]).coalesce(10)

os.system("hadoop fs -rm -r -skipTrash dev.db/" + str.lower(temp\_vl))

hiveContext.sql("drop table if exists dev."+temp\_vl)

temp\_vlt\_org.write.saveAsTable("dev."+temp\_vl)

## hiveContext写法 ##

order\_slice= hiveContext.table("app.app\_ rage").coalesce(10).where(col("dt")==self.order\_new\_dt).select("po\_no","goods\_no",col("distribution\_no").alias("dc\_id")).distinct().coalesce(10)

log =hiveContext.createDataFrame(\[\{"dt":dt,"types":types,"message":msg,"currtime":currTime\}\]).coalesce(1) \#types:"INFO" ,"ERROR"

hiveContext.sql("showpartitions app.app\_ rage").select(func.max("partition").alias("partition")).rdd.map(lambda x:str(x\[0\])).take(1)\[0\]

4.dataFrame数据写入hive表

temp\_vl\_org=self.vl\_data.join(self.order\_slice,\["po\_no"\]).coalesce(10)

os.system("hadoop fs -rm -r -skipTrash dev.db/" +str.lower(temp\_vl))

hiveContext.sql("drop table if exists dev."+temp\_vl)

temp\_vlt\_org.write.saveAsTable("dev."+temp\_vl)

## python 中执行hadoop命令 ##

\#删除表

os.system("hadoop fs -rm -r -skipTrashdev.db/" + str.lower(temp\_vl))

\#从集市1路径localFolde拷贝数据到集市2Hive表分区数据tbFolder

localFolder="app.db/app\_filter /"

partitonName="dt="+self.table\_new\_dt

tbFolder= "hdfs://102.1.1.1:8080/user/cm\_pc/app.db/app\_filter/"

os.system("hadoop fs -rm -r -skipTrash"+tbFolder+partitonName)

os.system("hadoop distcp"+localFolder+partitonName+" "+tbFolder+partitonName)

\# localFolder+partitonName为源路径，tbFolder+partitonName为目标路径

## dataFrame和rdd互转,存入Hive表 ##

\#以下返回dataFrame类型值

df = hc.sql(sql).coalesce(5)

\#dataFrame转rdd

r2 = df.rdd.map(lambda row :((row\[0\],row\[1\]),(row\[2\],row\[3\],row\[4\],row\[5\]))).groupByKey().map(lambda(k, v): sub\_process(k, v))

cond=\["rowkey","top\_10","dt"\]

\#rdd转dataFrame

result=r2.toDF(cond)

\#dataframe存入hive表

result.write.mode("append").insertInto(table\_name)

## 创建dataFrame方法 ##

法1：

\#rdd转dataFrame

result=r2.toDF(cond)

法2：

order\_slice= hiveContext.table("app.apprage").coalesce(10).where(col("dt")==self.ord\_new\_dt).select("po\_no","goods\_no",col("distribution\_no").alias("dc\_id")).distinct().coalesce(10)

法3：

log =hiveContext.createDataFrame(\[\{"dt":dt,"types":types,"message":msg,"currtime":currTime\}\]).coalesce(1) \#types:"INFO" ,"ERROR"

法4：

hiveContext.sql("showpartitions app.apprage")

法5：

从hdfs中读取数据变成dataFrame

input\_path为hdfs路径

textFile=sc.textFile(input\_path).map(lambdax: x.split("\\x01"))

df = textFile.toDF(\["seller\_no","seller\_name", "dept\_no", "goods\_no","shop\_id", "sp\_goods\_no", "dc\_id","erp\_warehouse\_no","real\_outstore\_qty", "sp\_create\_time"\]).groupby(\["seller\_no","seller\_name", "dept\_no", "goods\_no","dc\_id", "erp\_warehouse\_no","sp\_create\_time"\]).agg(func.sum("real\_outstore\_qty").alias("real\_outstore\_qty")).where((col("sp\_create\_time")< endDate) & (col("sp\_create\_time") >"2015-01-01"))

## pyspark when ##

dataDf = hiveContext.table(self.table\_name).where((col("dt")==self.table\_new\_dt)& (col("so\_status")<> "10028") &(col("so\_status")<> "10009"))\\

.select("seller\_no","seller\_name","dept\_no",

"goods\_no",col("shop\_id").cast("int"),F.when(func.isnull("sp\_goods\_no"),-1).otherwise(col("sp\_goods\_no").cast("int")).alias("sp\_goods\_no"),

col("distribution\_no").cast("int").alias("dc\_id"),col("erp\_warehouse\_no").cast("int"),

col("sp\_create\_time").cast("date"),F.when(func.isnull("apply\_out\_qty"),1).otherwise(col("apply\_out\_qty")).alias("apply\_qty"),).coalesce(10)

dataFrame添加列

法1,2

self.groupedDf=dataDf.groupBy("seller\_no","dept\_no","goods\_no","shop\_id","sp\_goods\_no","dc\_id","erp\_warehouse\_no","sp\_create\_time")\\

\#添加两列

.agg(func.max("seller\_name").alias("seller\_name"),func.sum("apply\_outstore\_qty").alias("apply\_outstore\_qty"))\\

.withColumn("dt",lit(self.table\_new\_dt))

## 判断是否执行成功 ##

if \_\_name\_\_ == "\_\_main\_\_":

main()

r = os.system("hadoop fs -test-e /tmp/for/result/e\_cast/\_SUCCESS")

if r != 0:

raise Exception("1")

else:

print ("执行spark获取eclp预测数据成功！！")

## 从hdfs导入数据到hive ##

def push2hive(self):

hiveContext.sql("LOAD DATA IN PATH '/tmp/for/result/ec\_forecast' OVERWRITE INTO TABLE app.app\_ ver1 PARTITION( dt  = '"+ \_today+"')")

## 读取全路径下的Hive表 ##

hdfs\_path\_ver1 ="/user/cmo\_ipc/app.db/app\_ ver1/dt=" + \_today

df1=sc.textFile(hdfs\_path\_ver1).map(lambda x:x.split("\\t")).map(lambdaline:((line\[0\],line\[1\],line\[2\],line\[3\]),line\[8\])).groupByKey().map(lambda(k,v): sales\_nation\_ver1(k,v))