• 从0开始学人工智能测试节选：Spark -- 结构化数据领域中测试人员的万金油技术（二）

    Dataframe

    dataframe 是spark中参考pandas设计出的一套高级API，用户可以像操作pandas一样方便的操作结构化数据。毕竟纯的RDD操作是十分原始且麻烦的。而dataframe的出现可以让熟悉pandas的从业人员能用非常少的成本完成分布式的数据分析工作， 毕竟跟数据打交道的人很少有不懂dataframe的。

    初始化dataframe的方法

    1. from pyspark import SparkContext, SparkConf, SQLContext
    2.  
    3. from pyspark.sql import Row
    4.  
    5.  
    6.  
    7. logFile = "/Users/xxxx/tools/spark-3.0.3-bin-hadoop2.7/README.md"
    8.  
    9. conf = SparkConf().setMaster("local").setAppName("My App")
    10.  
    11. sc = SparkContext(conf=conf)
    12.  
    13. sqlContext = SQLContext(sc)
    14.  
    15. dataA = sqlContext.read.csv("路径")
    16.  
    17.  
    18.  
    19.  
    20.  
    21. dicts = [{'col1': 'a', 'col2': 1}, {'col1': 'b', 'col2': 2}]
    22.  
    23. dataf = sqlContext.createDataFrame(dicts)
    24.  
    25. dataf.show()
    26.  
    27.  
    28.  
    29. dicts = [['a', 1], ['b', 2]]
    30.  
    31. rdd = sc.parallelize(dicts)
    32.  
    33. dataf = sqlContext.createDataFrame(rdd, ['col1','col2'])
    34.  
    35. dataf.show()
    36.  
    37.  
    38.  
    39.  
    40.  
    41. rows = [Row(col1='a', col2=1), Row(col1='b', col2=2)]
    42.  
    43. dataf= sqlContext.createDataFrame(rows)
    44.  
    45. dataf.show()
    46.  
    47.  
    48.  
    49. dataf.write.csv(path="/Users/cainsun/Downloads/test_spark", header=True, sep=",", mode='overwrite')

    可以看到创建dataframe有多种方式， 可以从文件中读取， 可以从列表中初始化，可以用简单的方式指定列信息， 也可以使用Row类来初始化列信息。

    dataframe常用操作

    读取数据：

    1. df = spark.read.json("data.json")
    2.  
    3.  
    4.  
    5. df = spark.read.csv("data.csv", header=True, inferSchema=True)
    6.  
    7.  
    8.  
    9. df = spark.read.parquet("data.parquet")

    显示数据:

    # 显示前 n 行数据，默认为 20 行

    df.show(n=5)

    # 打印 DataFrame 的 schema

    df.printSchema()

    选择和过滤数据:

    # 选择特定列

    selected_df = df.select("column1", "column2")

    # 使用条件过滤数据

    filtered_df = df.filter(df["age"] > 30)

    聚合和分组数据:

    1. from pyspark import SparkContext, SparkConf, SQLContext
    2.  
    3.  
    4.  
    5.  
    6.  
    7. conf = SparkConf().setMaster("local").setAppName("My App")
    8.  
    9. sc = SparkContext(conf=conf)
    10.  
    11. sqlContext = SQLContext(sc)
    12.  
    13.  
    14.  
    15.  
    16.  
    17. dicts = [
    18.  
    19. ['teacher', 202355, 16, '336051551@qq.com'],
    20.  
    21. ['student', 2035, 16, '336051551@qq.com'],
    22.  
    23. ['qa', 2355, 16, '336051551@qq.com'],
    24.  
    25. ['qa', 20235, 16, '336051551@qq.com'],
    26.  
    27. ['teacher', 35, 16, '336051asdf'],
    28.  
    29. ['student', 453, 16, '336051asdf'],
    30.  
    31.  
    32.  
    33.  
    34.  
    35. ]
    36.  
    37. rdd = sc.parallelize(dicts, 3)
    38.  
    39. data = sqlContext.createDataFrame(rdd, ['title', 'sales', 'age', 'email'])
    40.  
    41.  
    42.  
    43.  
    44.  
    45. result = data.groupBy("title").max("sales").alias("max_sales")
    46.  
    47. resultA = data.groupBy("title").sum("sales").alias("sum_sales")
    48.  
    49.  
    50.  
    51. # 显示结果
    52.  
    53. result.show()
    54.  
    55. resultA.show()
    56.  
    57.  
    58.  
    59.  
    60.  
    61. +-------+----------+
    62.  
    63. | title|max(sales)|
    64.  
    65. +-------+----------+
    66.  
    67. |teacher| 202355|
    68.  
    69. | qa| 20235|
    70.  
    71. |student| 2035|
    72.  
    73. +-------+----------+
    74.  
    75.  
    76.  
    77. +-------+----------+
    78.  
    79. | title|sum(sales)|
    80.  
    81. +-------+----------+
    82.  
    83. |teacher| 202390|
    84.  
    85. | qa| 22590|
    86.  
    87. |student| 2488|
    88.  
    89. +-------+----------+
    90.  
    91.  
    92.  
    93. 数据排序：
    94.  
    95.  
    96.  
    97. from pyspark.sql.functions import desc
    98.  
    99.  
    100.  
    101. # 按列排序
    102.  
    103. sorted_df = df.sort("column1")
    104.  
    105.  
    106.  
    107. # 按列降序排序
    108.  
    109. sorted_df = df.sort(desc("column1"))
    110.  
    111. 添加，修改和删除列：
    112.  
    113.  
    114.  
    115. from pyspark.sql.functions import upper
    116.  
    117.  
    118.  
    119. # 添加新列
    120.  
    121. new_df = df.withColumn("new_column", df["column1"] * 2)
    122.  
    123.  
    124.  
    125. # 修改现有列
    126.  
    127. modified_df = df.withColumn("column1", upper(df["column1"]))
    128.  
    129.  
    130.  
    131. # 删除列
    132.  
    133. dropped_df = df.drop("column1")
    134.  
    135.  
    136.  
    137. 重命名列:
    138.  
    139. # 重命名 DataFrame 中的列
    140.  
    141. renamed_df = df.withColumnRenamed("old_column_name", "new_column_name")

    spark sql

    初始化

    1. from pyspark import SparkContext, SparkConf, SQLContext
    2.  
    3.  
    4.  
    5. # 创建 SparkSession
    6.  
    7. conf = SparkConf().setMaster("local").setAppName("My App")
    8.  
    9. sc = SparkContext(conf=conf)
    10.  
    11. sqlContext = SQLContext(sc)
    12.  
    13.  
    14.  
    15.  
    16.  
    17. dicts = [
    18.  
    19. ['teacher', 202355, 16, '336051551@qq.com'],
    20.  
    21. ['student', 2035, 16, '336051551@qq.com'],
    22.  
    23. ['qa', 2355, 16, '336051551@qq.com'],
    24.  
    25. ['qa', 20235, 16, '336051551@qq.com'],
    26.  
    27. ['teacher', 35, 16, '336051asdf'],
    28.  
    29. ['student', 453, 16, '336051asdf'],
    30.  
    31.  
    32.  
    33.  
    34.  
    35. ]
    36.  
    37. rdd = sc.parallelize(dicts, 3)
    38.  
    39. data = sqlContext.createDataFrame(rdd, ['title', 'sales', 'age', 'email'])
    40.  
    41.  
    42.  
    43. data.createOrReplaceTempView("table")

    要使用spark sql的能力， 需要利用createOrReplaceTempView创建一个临时表，然后才能执行 sql

    简单的sql执行

    1. query = "select * from table where title = 'qa'"
    2.  
    3.  
    4.  
    5. resultB = sqlContext.sql(query)
    6.  
    7.  
    8.  
    9. resultB.show()
    10.  
    11.  
    12.  
    13. # 执行结果
    14.  
    15. +-----+-----+---+----------------+
    16.  
    17. |title|sales|age| email|
    18.  
    19. +-----+-----+---+----------------+
    20.  
    21. | qa| 2355| 16|336051551@qq.com|
    22.  
    23. | qa|20235| 16|336051551@qq.com|
    24.  
    25. +-----+-----+---+----------------+

    分组查询

    1. query = "select title, sum(sales), max(sales) from table group by title"
    2.  
    3.  
    4.  
    5. resultC = sqlContext.sql(query)
    6.  
    7.  
    8.  
    9. resultC.show()
    10.  
    11.  
    12.  
    13. # 执行结果
    14.  
    15. +-------+----------+----------+
    16.  
    17. | title|sum(sales)|max(sales)|
    18.  
    19. +-------+----------+----------+
    20.  
    21. |teacher| 202390| 202355|
    22.  
    23. | qa| 22590| 20235|
    24.  
    25. |student| 2488| 2035|
    26.  
    27. +-------+----------+----------+

    Spark sql适合熟悉sql语法的人使用，本质上sql和dataframe最终都会被翻译成rdd来运行。我们可以把它看成是rdd的高级语法糖就可以。 大家喜欢哪种操作方式就选择哪种就可以。

    数据测试/监控

    回顾自学习与数据闭环那里，我们知道在这样的系统中针对与每天新采集的数据，需要做一道数据校验。下面我模拟一个场景写这样一个检查脚本。

    1. from pyspark import SparkContext, SparkConf, SQLContext
    2.  
    3. from pyspark.sql import SparkSession
    4.  
    5. import pyspark.sql.functions as F
    6.  
    7.  
    8.  
    9. conf = SparkConf().setMaster("local").setAppName("My App")
    10.  
    11. sc = SparkContext(conf=conf)
    12.  
    13. sqlContext = SQLContext(sc)
    14.  
    15.  
    16.  
    17. rdd = sc.parallelize(range(1000))
    18.  
    19. print(rdd.map(lambda x: '%s,%s' % ('男', '16')).collect())
    20.  
    21.  
    22.  
    23. dicts = [
    24.  
    25. ['frank', 202355, 16, '336051551@qq.com'],
    26.  
    27. ['frank', 202355, 16, '336051551@qq.com'],
    28.  
    29. ['frank', 202355, 16, '336051551@qq.com'],
    30.  
    31. ['frank', 202355, 16, '336051551@qq.com'],
    32.  
    33. ['frank', 202355, 16, '336051asdf'],
    34.  
    35. ['', 452345, 16, '336051asdf'],
    36.  
    37.  
    38.  
    39.  
    40.  
    41. ]
    42.  
    43. rdd = sc.parallelize(dicts, 3)
    44.  
    45. dataf = sqlContext.createDataFrame(rdd, ['name', 'id', 'age', 'email'])
    46.  
    47.  
    48.  
    49.  
    50.  
    51. # 验证 id 字段必须是整数
    52.  
    53. id_filter = F.col("id").cast("int") >= 0
    54.  
    55.  
    56.  
    57. # 验证 name 字段必须是非空字符串
    58.  
    59. name_filter = F.col("name").isNotNull() & (F.col("name") != "")
    60.  
    61.  
    62.  
    63. # 验证 age 字段必须是大于等于 0 的整数
    64.  
    65. age_filter = F.col("age").cast("int") >= 0
    66.  
    67.  
    68.  
    69. # 验证 email 字段必须是有效的电子邮件地址
    70.  
    71. email_filter = F.col("email").rlike("^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$")
    72.  
    73.  
    74.  
    75. # 应用过滤条件
    76.  
    77. valid_data = dataf.filter(id_filter & name_filter & age_filter & email_filter)
    78.  
    79.  
    80.  
    81. # 输出符合质量要求的数据
    82.  
    83. valid_data.show()
    84.  
    85.  
    86.  
    87. # 输出不符合质量要求的数据
    88.  
    89. invalid_data = dataf.exceptAll(valid_data)
    90.  
    91. invalid_data.show()

    更多内容欢迎来到我的知识星球：
     

  • 相关阅读:
    MVC使用的设计模式
    mybatis源码阅读系列（一）
    外包干了一个月，技术明显进步。。。。。
    日期分析处理
    Basis运维日常检查清单- Checklist
    智能驾驶ADAS算法设计及Prescan仿真(3): 自适应巡航ACC跟车目标选择策略设计与simulink仿真
    mysql数据库安装
    Python人员信息管理系统（简直期末人福音）
    DTO的作用
    CentOS shell中的变量
  • 原文地址：https://blog.csdn.net/ycwdaaaa/article/details/137912207