PySpark Basics: Graphing可视化
有很多方法可以汇总图表的数据（例如，六角形，方块图，条形图），当数据很大时，聚合减小了大小。如果聚合足以让你的数据被加载到内存中，那么没有问题;你可以使用任何你喜欢的绘图工具。您可以将数据下载到S3，也可以从本地存储到您的计算机上，然后按照这种方式进行绘制。您也可以使用Python提供的绘图工具，我们将在这里进行介绍。

绘制大数据而不汇总（例如在二元散点图中）会随着数据变大而变得困难。有两种可能的解决方案：首先，可以使用Amazon Web Services来启动具有大量内存的单台机器，以创建图形的唯一目的。机器本身可能是昂贵的访问，但这可能会被抵消，只需要非常简短。目前还没有一个适合这种情况的系统。第二个选项在Spark中还没有经过测试，那就是使用Python的Bokeh DataShader模块。是否在分布式环境中工作将成为未来工作的主题。

import pandas as pd
import matplotlib.pyplot as plt
%matplotlib inline              #tells the Jupyter Notebook to display graphs inline (rather than in a separate window)

1 Loading and Viewing the Data

spark_df = spark.read.csv('s3://ui-spark-social-science-public/data/diamonds.csv', inferSchema=True, header=True, sep=',')
spark_df.show(5)
+-----+-------+-----+-------+-----+-----+-----+----+----+----+
|carat|    cut|color|clarity|depth|table|price|   x|   y|   z|
+-----+-------+-----+-------+-----+-----+-----+----+----+----+
| 0.23|  Ideal|    E|    SI2| 61.5| 55.0|  326|3.95|3.98|2.43|
| 0.21|Premium|    E|    SI1| 59.8| 61.0|  326|3.89|3.84|2.31|
| 0.23|   Good|    E|    VS1| 56.9| 65.0|  327|4.05|4.07|2.31|
| 0.29|Premium|    I|    VS2| 62.4| 58.0|  334| 4.2|4.23|2.63|
| 0.31|   Good|    J|    SI2| 63.3| 58.0|  335|4.34|4.35|2.75|
+-----+-------+-----+-------+-----+-----+-----+----+----+----+

spark_df.dtypes
[('carat', 'double'),
 ('cut', 'string'),
 ('color', 'string'),
 ('clarity', 'string'),
 ('depth', 'double'),
 ('table', 'double'),
 ('price', 'int'),
 ('x', 'double'),
 ('y', 'double'),
 ('z', 'double')]

spark_df.describe(['carat', 'depth', 'table', 'price']).show()
+-------+------------------+------------------+------------------+-----------------+
|summary|             carat|             depth|             table|            price|
+-------+------------------+------------------+------------------+-----------------+
|  count|             53940|             53940|             53940|            53940|
|   mean|0.7979397478679852| 61.74940489432624| 57.45718390804603|3932.799721913237|
| stddev|0.4740112444054196|1.4326213188336525|2.2344905628213247|3989.439738146397|
|    min|               0.2|              43.0|              43.0|              326|
|    max|              5.01|              79.0|              95.0|            18823|
+-------+------------------+------------------+------------------+-----------------+

2 Using Matplotlib

Python中有很多图形解决方案，但最常见的是Matplotlib。 这是非常灵活的; 从字面上看，如果需要，图表的每个方面都可以被操纵和微调。 这可以使它成为一个很好的工具，但同样的灵活性往往会导致复杂而冗长的代码。

我们将从一个非常简单的例子开始，它只使用Matplotlib可以做的最少的事情，然后让它为我们填充默认设置。 请记住，为了在Matplotlib中绘制图表，我们的数据不能分发; 它必须足够小才能收集。 因此，我们首先从Spark数据框中提取要直接绘制的两列的值：

carat = spark_df[['carat']].collect()
price = spark_df[['price']].collect()
print(carat[:5])
print(price[:5])
[Row(carat=0.23), Row(carat=0.21), Row(carat=0.23), Row(carat=0.29), Row(carat=0.31)]
[Row(price=326), Row(price=326), Row(price=327), Row(price=334), Row(price=335)]

fig = plt.figure()
ax = fig.add_subplot(1,1,1)

ax.plot(carat, price, 'go', alpha=0.1)
ax.set_xlabel('Carat')
ax.set_ylabel('Price')
ax.set_title('Diamonds')

image.png

#fig = plt.figure()
#ax1 = fig.add_subplot(2,2,1) #plot 1
#ax2 = fig.add_subplot(2,2,2) #plot 2
#ax3 = fig.add_subplot(2,2,3) #plot 3
#ax4 = fig.add_subplot(2,2,4) #plot 4
#fig.suptitle('Title for the Figure')
#ax1.set_title('Title for the First Subplot')

3 Using Pandas

如果您首先将Spark数据框转换为Pandas数据框，则可以简化Python中的绘图。 除了Pandas方便的绘图工具之外，还可以访问许多其他与R语言类似的数据框操作。 再次注意，您只能在足够小的数据集上使用Panads，以适应一个系统; 像本地R数据帧一样，它们不支持分布式环境。 因此，您可能需要将大型分布式Spark数据集分组，然后才能加载并以此方式工作。 与Matplotlib一样，Pandas可以轻松处理我们的钻石数据集。

R，Pandas，PySpark和SparkR都有共同的血统，虽然有不同之处，但当他们之间存在差异的时候，还有很多相似之处，可以缓解来回的过渡。

Spark数据框有一个内置的方法来把自己变成Pandas数据框：

pandas_df = spark_df.toPandas()
pandas_df.describe()

carat   depth   table   price   x   y   z
count   53940.000000    53940.000000    53940.000000    53940.000000    53940.000000    53940.000000    53940.000000
mean    0.797940    61.749405   57.457184   3932.799722 5.731157    5.734526    3.538734
std 0.474011    1.432621    2.234491    3989.439738 1.121761    1.142135    0.705699
min 0.200000    43.000000   43.000000   326.000000  0.000000    0.000000    0.000000
25% 0.400000    61.000000   56.000000   950.000000  4.710000    4.720000    2.910000
50% 0.700000    61.800000   57.000000   2401.000000 5.700000    5.710000    3.530000
75% 1.040000    62.500000   59.000000   5324.250000 6.540000    6.540000    4.040000
max 5.010000    79.000000   95.000000   18823.000000    10.740000   58.900000   31.800000

请注意，我们将reset_index添加到最后，因为我们需要将 clarity 视为列而不是索引，并且默认情况下，Pandas使groupby类别成为索引。 PySpark数据框没有标记，但Pandas数据框不能没有; 当我们把 clarity再次列为一个列而不是一个索引时，Pandas只是默认填充从0开始的整数索引。

grouped = pandas_df.groupby('clarity').agg({'price':'mean'}).reset_index()
clarity_order = ['FL', 'IF', 'VVS1', 'VVS2', 'VS1', 'VS2', 'SI1', 'SI2', 'I1', 'I2', 'I3']
mapping = {day: i for i, day in enumerate(clarity_order)}
key = grouped['clarity'].map(mapping)
grouped = grouped.iloc[key.argsort()]
grouped.plot(kind='bar', x='clarity', legend=False)

image.png

4 Pandas and Matplotlib Together

ax = grouped.plot(kind='bar', x='clarity', legend=False)
ax.set_xlabel('Clarity')
ax.set_ylabel('Price')
ax.set_title('Diamonds')

image.png