pyspark处理数据基本语法

import findspark #初始化 findspark.init() import warnings warnings.filterwarnings(ignore) from pyspark.sql import SparkSession # 定义数据库的地址，以及表，登录用户及密码 url = “jdbc:mysql://localhost:3306/xx” table=“table” #密码账户需要字典的形式传入 properties ={“user”:“root”,“password”:“12345678”} spark = SparkSession.builder.appName(My first app).getOrCreate() df = spark.read.jdbc(url=url,table=table,properties=properties) df.show(4)

df.count(),len(df.columns)# 查看数据维度

df.printSchema()# 元数据分析，查看字段类型

df.select([cxkh,jymc]) # 按照列名选择

df.filter((df[jyje]>100000) & (df[jysj]>2015-01-01)) #按照条件进行筛选

#在元数据上更改字段 df.withColumn(“jyje”,df3.jyje.astype(“int”))

df3.describe().show()

df.withColumn(test,(df[jyje]+100))

df[df[jydfmc].isNull()] # 筛选未空

dd.fillna(wz).show(2) #缺失值填充 dd.na.fill(wz).show(2) #缺失值填充

#按照一个index分组 df.groupBy(jymc).count().orderBy(count,ascending=False).show(3) #按照两个index分组 df.groupBy(jymc,cxkh).agg({jyje:sum,jysj:max}).orderBy(sum(jyje),ascending=False).show(5,False) #特定函数聚合求解

df.select([cxkh,jymc]).dropDuplicates().show(5)

from pyspark.sql.functions import udf from pyspark.sql.types import StringType #类似于ｌａｍｂｄａ brand_udf=udf(jyje_level,StringType()) #注册ｕｄf df.withColumn(jyje_level,brand_udf(df[jyje])).show(5)

df.select([cxkh,jymc,jdbz,jyje,jydfmc]).filter(“jydfmc rlike 基金|理财|证券”) # 模糊匹配

df.select(cxkh,jysj,jyje).groupBy(cxkh).agg({jysj:max,jyje:max}).\ withColumnRenamed(cxkh,cxkh1).withColumnRenamed(max(jysj),最晚一次交易时间).\ withColumnRenamed(max(jyje),最大交易金额) #修改指定字段的名字， df.seleceExpr(first_name as lase_name),这个好像需要字段名全部写上，否则最终显示结果只有更改了列名的那一列。

df_person.join(df_lastime,df_person[cxkh]==df_lastime[cxkh1],\ left_outer).orderBy(cxkh,ascending=True)# 两表拼接指定字段的拼接，其他的连接方式：inner, outer, left_outer, right_outer, leftsemi

df.registerTempTable(df1) # 将数据框注册为一个table spark.sql(“select * from df1where jymc=xxx”).show(2)#程序 mysql程序

df.select(jymc).dropDuplicates().intersect(df.select(jydfmc).dropDuplicates())#交集 df.select(jymc).dropDuplicates().subtract(df.select(jydfmc).dropDuplicates())#差集 df.select(jymc).dropDuplicates().union(df.select(jydfmc).dropDuplicates())#并集

df.select(jymc).union(df.select(jydfmc)).distinct()#并集+去重

from pyspark.sql.functions import split df.withColumn(jyrq,split(df_year[jysj],” “).getItem(0))#拆分后取第一个元素 df.withColumn(jytime,split(df_year[jysj],” “)[1])#拆分后取第二个元素

from pyspark.sql.functions import month,year,dayofmonth,dayofweek,dayofyear df.withColumn(year,year(jysj)).\ withColumn(month,month(jysj)).\ withColumn(day,dayofmonth(jysj)).\ withColumn(week,dayofweek(jysj)).\ withColumn(day_num,dayofyear(jysj)) # 获取对应的年，月，日，一周内第几天，一年内第几天

#写出在多个csv文件中，一个csv文件为一行数据 df.write.save(path.csv) #写出在一个csv 文件中 df.repartition(1).write.csv(“path.csv”,header=True, sep=”,”, mode=overwrite,encoding=utf-8) #保存为parquet格式数据。读取速度快，占用内存小 df_person_info_lasttime.write.format(parquet).save(path4.parquet) #读取parquet数据 d1=spark.read.format(parquet).load(path.csv4.parquet)