Merge
1.Stacking Rows with Matching Columns(行堆积)
你可能在每个数据框中都有相同的列，只是想将其中一个堆叠在另一个之上。
from pyspark.sql import Row
row = Row("name", "pet", "count")
df1 = sc.parallelize([
row("Sue", "cat", 16),
row("Kim", "dog", 1),
row("Bob", "fish", 5)
]).toDF()
df2 = sc.parallelize([
row("Fred", "cat", 2),
row("Kate", "ant", 179),
row("Marc", "lizard", 5)
]).toDF()
df3 = sc.parallelize([
row("Sarah", "shark", 3),
row("Jason", "kids", 2),
row("Scott", "squirrel", 1)
]).toDF()
df_union = df1.unionAll(df2)
df_union.show()
+----+------+-----+
|name| pet|count|
+----+------+-----+
| Sue| cat| 16|
| Kim| dog| 1|
| Bob| fish| 5|
|Fred| cat| 2|
|Kate| ant| 179|
|Marc|lizard| 5|
+----+------+-----+
unionAll方法只允许我们一次堆栈两个数据帧。 如果有多个数据帧以这种方式堆叠，我们可以重复这样做，但是我们也可以使用一个辅助函数UDF来使它更容易。
标准的Python命令reduce将一个函数应用到项目列表中，以便将其“减少”到一个输出。 有了这个，你可以将任意数量的数据框传递给我们的UDF函数，它们会堆积在一起：
from pyspark.sql import DataFrame
from functools import reduce
def union_many(*dfs):
return reduce(DataFrame.unionAll, dfs)
df_union = union_many(df1, df2, df3)
df_union.show()
+-----+--------+-----+
| name| pet|count|
+-----+--------+-----+
| Sue| cat| 16|
| Kim| dog| 1|
| Bob| fish| 5|
| Fred| cat| 2|
| Kate| ant| 179|
| Marc| lizard| 5|
|Sarah| shark| 3|
|Jason| kids| 2|
|Scott|squirrel| 1|
+-----+--------+-----+
2.Merging Columns by Matching Rows
合并的另一种方式是通过跨行的某些键组合列。
在我们构建数据之后，有四种方法可以指定操作的逻辑：
row1 = Row("name", "pet", "count")
row2 = Row("name", "pet2", "count2")
df1 = sc.parallelize([
row1("Sue", "cat", 16),
row1("Kim", "dog", 1),
row1("Bob", "fish", 5),
row1("Libuse", "horse", 1)
]).toDF()
df2 = sc.parallelize([
row2("Sue", "eagle", 2),
row2("Kim", "ant", 179),
row2("Bob", "lizard", 5),
row2("Ferdinand", "bees", 23)
]).toDF()
首先我们要做一个“内连接”（inner join），它将在两个数据框中都合并的行和在其他所有行中合并的行。 这是默认的连接类型，所以如果你不希望显式地指明（如果显式几乎总是更好的话），那么how参数可以省略。 我们将在name列中的条目进行合并，您可以看到这是方法中的第二个参数; 如果合并发生在多个匹配值上，这也可以是一个list：
df1.join(df2, 'name', how='inner').show()
+----+----+-----+------+------+
|name| pet|count| pet2|count2|
+----+----+-----+------+------+
| Sue| cat| 16| eagle| 2|
| Bob|fish| 5|lizard| 5|
| Kim| dog| 1| ant| 179|
+----+----+-----+------+------+
df1.join(df2, 'name', how='outer').show()
df1.join(df2, 'name', how='left').show()
Missing-data
1 探索Null值
df = spark.read.csv('s3://ui-spark-social-science-public/data/Performance_2015Q1.txt',
header=False, inferSchema=True, sep='|', nullValue='')
请注意，在nullValue=''行中，空字符串可以被数据集使用的任何值替换 - 这是告诉Spark哪些值应该在数据框中转换为空值。
df.count()
3526154
df.dtypes
[('_c0', 'bigint'),
('_c1', 'string'),
('_c2', 'string'),
('_c3', 'double'),
('_c4', 'double'),
('_c5', 'int'),
('_c6', 'int'),
('_c7', 'int'),
('_c8', 'string'),
('_c9', 'int'),
('_c10', 'string'),
('_c11', 'string'),
('_c12', 'int'),
('_c13', 'string'),
('_c14', 'string'),
('_c15', 'string'),
('_c16', 'string'),
('_c17', 'string'),
('_c18', 'string'),
('_c19', 'string'),
('_c20', 'string'),
('_c21', 'string'),
('_c22', 'string'),
('_c23', 'string'),
('_c24', 'string'),
('_c25', 'string'),
('_c26', 'int'),
('_c27', 'string')]
df.where( df['_c12'].isNull() ).count()
3510294
只能在列上调用isNull（）方法，而不是在整个df上调用
利用UDF函数循环每列超出每列的空值个数
def count_nulls(df):
null_counts = []
for col in df.dtypes:
cname = col[0]
ctype = col[1]
if ctype != 'string':
nulls = df.where( df[cname].isNull() ).count()
result = tuple([cname, nulls])
null_counts.append(result)
return null_counts
null_counts = count_nulls(df)
null_counts
[('_c0', 0),
('_c3', 0),
('_c4', 1945752),
('_c5', 0),
('_c6', 0),
('_c7', 1),
('_c9', 0),
('_c12', 3510294),
('_c26', 3526153)]
2 删除空值
现在我们可以对'null'值做三件事，即我们知道数据框中的内容。
We can ignore them, we can drop them, or we can replace them。
请记住，PySpark数据框是不可变的，所以我们实际上不能改变原始数据集。 所有的操作都会返回一个全新的数据框，尽管我们可以用df = df.some_operation（）来覆盖现有的数据框。
df_drops = df.dropna(how='all', subset=['_c4', '_c12', '_c26'])
df_drops.count()
df.dropna（）方法在这里有两个参数：'how'可以等于'any'或者'all' 如果any的值为null。
subset参数需要一个你想查找null值的列的列表。 它实际上并不是数据框的子集; 它只是检查这三列，然后删除整个数据帧的行，如果该子集满足条件。 如果它应该检查所有的列为null，这可以被关闭。
“dropna（）”可以使用第三个参数。 在“thresh”参数设置一个阈值之前，一个行中的“空”条目将其丢弃。 它被设置为一个整数，用于指定该行必须具有多少个非空参数。 如果它小于这个数字就会丢弃该行。**如果你像下面那样指定了这个参数，它将返回一个数据帧，其中指定子集中少于2个非空值的行被删除：
所以我们可以在上面看到，一旦我们删除'_c4，_c12和_c26列中所有具有null`值的行，我们在原来的3,526,154行中就剩下1,580,403行 在整个数据帧上。
df_drops2 = df.dropna(thresh=2, subset=['_c4', '_c12', '_c26'])
df_drops2.count()
3 替换空值
df_fill = df.fillna(0, subset=['_c12'])
df_fill.where( df_fill['_c12'].isNull() ).count()
0
我们看到它取代了之前发现的所有3,510,294个空值。 “fillna（）”中的第一项可以是任何类型和任何值，如果填充应用于所有列，子集列表可以保持不变（尽管确保dtype与已经在该列中的内容一致）。 请注意，df.replace（a，b）做同样的事情，只有你指定a作为被替换的值，b作为替换。 它也接受可选的子集列表，但不利用优化的空值处理。
网友评论