什么是SPARK?

　　1. 先进的大数据分布式编程和计算框架

　　2. 替换Hadoop 中的MR计算引擎。

　　3. 内存分布式计算:运行数度快

　　4. 可以使用不同的语言编程(java,scala,r 和python)

　　5. 可以从不同的数据源获取数据，可以从HDFS,Cassandea,HBase等等，同时可以支持很多的文件格式:text Seq AVRO Parquet

　　6. 实现不同的大数据功能:Spark Core,Sparc SQL等等

　　PySpark 是 Spark 为 Python 开发者提供的 API ，位于 $SPARK_HOME/bin 目录，其依赖于 Py4J。随着Spark 2.1.0发布的 Py4J位于 $SPARK_HOME/python/lib 目录，对应的版本是 0.10.4。

# Imports

from pyspark.sql import SparkSession

# Create SparkSession

spark = SparkSession.builder \

    .appName('SparkByExamples.com') \

    .config("spark.jars", "mysql-connector-java-5.1.28.jar") \

    .getOrCreate()

# Read from MySQL Table

table_df = spark.read \

    .format("jdbc") \

    .option("driver", "com.mysql.jdbc.Driver") \

    .option("url", "jdbc:mysql://134.**.**.**:9200/hesc_stm_xhm") \

    .option("dbtable", "temp_user_grid") \

    .option("user", "root") \

    .option("password", "****") \

    .load()

# check  read accessable

# print( table_df.count())  # 总行数

# etl 使用rdd 算子

rdd = table_df.rdd

# print(rdd.first())

#  cityCode

# print(rdd.filter(lambda r: r(5) == None).count())  # gridCode为空的行数

rdd1 = rdd.filter(lambda r: Row.asDict(r).get("cityCode") != None).filter(

    lambda r: len(Row.asDict(r).get("cityCode")) == 9)

# print(rdd.map(lambda r: Row.asDict(r).get("cityCode")).take(5))   # ROW类型的元素读取 使用 r(19)读取列有问题

def checkCityCode(str):

    # 判断字符串的格式，前3位为001，而且全为数字

    if (str[:3] == '001') and str.isnumeric():

        return True

    else:

        return False

rdd2 = rdd1.filter(lambda r: checkCityCode(Row.asDict(r).get("cityCode")))

print(rdd2.first())

#  数据分析 使用 rdd  df算子 sql 三种算子 ; 统计不同网格的人员数量。

# rdd operator

map = rdd2.map(lambda r: (Row.asDict(r).get("gridCode"), Row.asDict(r).get("id"))).countByKey()

print(map)   #  查询python  rdd api

# df/ds operator   dataset 1.6之后加入， 整合了RDD 的强类型便于使用lambda函数以及 sqpark sql 优化引擎

# python 没有dataset 类型。java scala 可以。 dataframe是 dataset 的 一种。 dataframe 适用python .

df = rdd2.toDF()

df1 = df.groupBy('gridCode').count()  # dataframe  特定编程语言 对结构化数据操作， 也称 无类型dataset算子

df1.show(4)

# sql  operator

df.createOrReplaceTempView('temp_user_grip')

df2 = spark.sql("select gridCode, count(id)  from temp_user_grip group by gridCode")

df2.show(2)

spark.stop()