商务网站运营与管理,赶集网的二级域名网站怎么做,国外教做美食网站,网站文件目录2、DataFrame
2.1 介绍
在Spark语义中#xff0c;DataFrame是一个分布式的行集合#xff0c;可以想象为一个关系型数据库的表#xff0c;或者一个带有列名的Excel表格。它和RDD一样#xff0c;有这样一些特点#xff1a;
Immuatable#xff1a;一旦RDD、DataFrame被创…2、DataFrame
2.1 介绍
在Spark语义中DataFrame是一个分布式的行集合可以想象为一个关系型数据库的表或者一个带有列名的Excel表格。它和RDD一样有这样一些特点
Immuatable一旦RDD、DataFrame被创建就不能更改只能通过transformation生成新的RDD、DataFrameLazy Evaluations只有action才会触发Transformation的执行DistributedDataFrame和RDD一样都是分布式的dataframe和dataset统一dataframe只是dataset[ROW]的类型别名。由于Python是弱类型语言只能使用DataFrame
DataFrame vs RDD
RDD分布式的对象的集合Spark并不知道对象的详细模式信息DataFrame分布式的Row对象的集合其提供了由列组成的详细模式信息使得Spark SQL可以进行某些形式的执行优化。DataFrame和普通的RDD的逻辑框架区别如下所示 左侧的RDD Spark框架本身不了解 Person类的内部结构。 右侧的DataFrame提供了详细的结构信息schema——每列的名称类型 DataFrame还配套了新的操作数据的方法DataFrame API如df.select())和SQL(select id, name from xx_table where …)。 DataFrame还引入了off-heap,意味着JVM堆以外的内存, 这些内存直接受操作系统管理而不是JVM。 RDD是分布式的Java对象的集合。DataFrame是分布式的Row对象的集合。DataFrame除了提供了比RDD更丰富的算子以外更重要的特点是提升执行效率、减少数据读取以及执行计划的优化。 DataFrame的抽象后我们处理数据更加简单了甚至可以用SQL来处理数据了 通过DataFrame API或SQL处理数据会自动经过Spark 优化器Catalyst的优化即使你写的程序或SQL不高效也可以运行的很快。 DataFrame相当于是一个带着schema的RDD
Pandas DataFrame vs Spark DataFrame
Cluster Parallel集群并行执行Lazy Evaluations: 只有action才会触发Transformation的执行Immutable不可更改Pandas rich API比Spark SQL api丰富
2.2 创建DataFrame
1创建dataFrame的步骤
调用方法例如spark.read.xxx方法
2其他方式创建dataframe createDataFramepandas dataframe、list、RDD 数据源RDD、csv、json、parquet、orc、jdbc jsonDF spark.read.json(xxx.json)jsonDF spark.read.format(json).load(xxx.json)parquetDF spark.read.parquet(xxx.parquet)jdbcDF spark.read.format(jdbc).option(url,jdbc:mysql://localhost:3306/db_name).option(dbtable,table_name).option(user,xxx).option(password,xxx).load()Transformation:延迟性操作 action立即操作
2.3 DataFrame API实现
基于RDD创建
from pyspark.sql import SparkSession
from pyspark.sql import Rowspark SparkSession.builder.appName(test).getOrCreate()
sc spark.sparkContext
# spark.conf.set(spark.sql.shuffle.partitions, 6)
# 直接创建
l [(Ankit,25),(Jalfaizy,22),(saurabh,20),(Bala,26)]
rdd sc.parallelize(l)
#为数据添加列名
people rdd.map(lambda x: Row(namex[0], ageint(x[1])))
#创建DataFrame
schemaPeople spark.createDataFrame(people)从csv中读取数据
# 从csv读取
#加载csv类型的数据并转换为DataFrame
df spark.read.format(csv). \option(header, true) \.load(iris.csv)
#显示数据结构
df.printSchema()
#显示前10条数据
df.show(10)
#统计总量
df.count()
#列名
df.columns增加一列
# 增加一列(或者替换) withColumn
#定义一个新的列数据为其他某列数据的两倍
#如果操作的是原有列可以替换原有列的数据
df.withColumn(newWidth,df.SepalWidth * 2).show()删除一列
# 删除一列 drop
#删除一列
df.drop(cls).show()统计信息
# 统计信息 describe
df.describe().show()
#计算某一列的描述信息
df.describe(cls).show() 提取部分列
# 提取部分列 select
df.select(SepalLength,SepalWidth).show()基本统计功能
# 基本统计功能 distinct count
df.select(cls).distinct().count()分组统计
# 分组统计 groupby(colname).agg({col:fun,col2:fun2})
df.groupby(cls).agg({SepalWidth:mean,SepalLength:max}).show()# avg(), count(), countDistinct(), first(), kurtosis(),
# max(), mean(), min(), skewness(), stddev(), stddev_pop(),
# stddev_samp(), sum(), sumDistinct(), var_pop(), var_samp() and variance()自定义的汇总方法
# 自定义的汇总方法
import pyspark.sql.functions as fn
#调用函数并起一个别名
df.agg(fn.count(SepalWidth).alias(width_count),fn.countDistinct(cls).alias(distinct_cls_count)).show()拆分数据集
#数据集拆成两部分 randomSplit
#设置数据比例将数据划分为两部分
trainDF, testDF df.randomSplit([0.6, 0.4])采样数据
# 采样数据 sample
#withReplacement是否有放回的采样
#fraction采样比例
#seed随机种子
sdf df.sample(False,0.2,100)查看两个数据集在类别上的差异
#查看两个数据集在类别上的差异 subtract确保训练数据集覆盖了所有分类
diff_in_train_test testDF.select(cls).subtract(trainDF.select(cls))
diff_in_train_test.distinct().count()交叉表
# 交叉表 crosstab
df.crosstab(cls,SepalLength).show()udf
udf自定义函数
# 综合案例 udf
# 测试数据集中有些类别在训练集中是不存在的找到这些数据集做后续处理
trainDF,testDF df.randomSplit([0.99,0.01])diff_in_train_test trainDF.select(cls).subtract(testDF.select(cls)).distinct().show()#首先找到这些类整理到一个列表
not_exist_cls trainDF.select(cls).subtract(testDF.select(cls)).distinct().rdd.map(lambda x :x[0]).collect()#定义一个方法用于检测
def should_remove(x):if x in not_exist_cls:return -1else :return x#创建udfudf函数需要两个参数
# Function
# Return type (in my case StringType())#在RDD中可以直接定义函数交给rdd的transformatioins方法进行执行
#在DataFrame中需要通过udf将自定义函数封装成udf函数再交给DataFrame进行调用执行from pyspark.sql.types import StringType
from pyspark.sql.functions import udfcheck udf(should_remove,StringType())resultDF trainDF.withColumn(New_cls,check(trainDF[cls])).filter(New_cls -1)resultDF.show()
