1. 从dataframe arraytype列获取前n个元素

有一个Spark dataframe ，其中的行为

  1. df.show()
  2. +---+---------+
  3. | id|  letters|
  4. +---+---------+
  5. |  1|[a, b, c]|
  6. |  2|[d, e, f]|
  7. |  3|[g, h, i]|
  8. +---+---------+
  12. # 记住这个东西，数据类型很重要，用做定位、分类针对性解决问题
  13. df.printSchema()
  14. |-- id: long (nullable = true)
  15. |-- letters: array (nullable = true)
  16. |    |-- element: string (containsNull = true)

只想保留 array列的前2个元素

  1. df.show()
  2. +---+---------+
  3. | id|  letters|
  4. +---+-----+
  5. |  1|[a, b]|
  6. |  2|[d, e]|
  7. |  3|[g, h]|
  8. +---+------+

操作：
可以使用方括号按索引访问Letters列中的元素，并将其包装在对pyspark.sql.functions.array（）的调用中以创建新的arrayType列。

  1. import pyspark.sql.functions as f
  3. df.withColumn("first_two", f.array([f.col("letters")[0], f.col("letters")[1]])).show()
  4. +---+---------+---------+
  5. | id|  letters|first_two|
  6. +---+---------+---------+
  7. |  1|[a, b, c]|   [a, b]|
  8. |  2|[d, e, f]|   [d, e]|
  9. |  3|[g, h, i]|   [g, h]|
  10. +---+---------+---------+
  13. # 或者如果有太多的索引要列出，可以使用列表理解：
  14. df.withColumn("first_two", f.array([f.col("letters")[i] for i in range(2)])).show()
  15. +---+---------+---------+
  16. | id|  letters|first_two|
  17. +---+---------+---------+
  18. |  1|[a, b, c]|   [a, b]|
  19. |  2|[d, e, f]|   [d, e]|
  20. |  3|[g, h, i]|   [g, h]|
  21. +---+---------+---------+

2. 将一列string转为一列array

这里用的是分词

  1. #原DataFrame
  2. col1 | my_str_col
  3. ----- +-----------
  4.   1    | qxl《电力设备》杂志
  5.   2    |  qxl《电力设备》杂志
  6. # 期望结果 1
  7. col1 | my_str_col
  8. ----- +-----------
  9.   1    |  [q, x, l,  , 电, 力, 设, 备,  , 杂, 志]
  10.   2    |  [q, x, l,  , 电, 力, 设, 备,  , 杂, 志]
  12. # 期望结果 2
  13. col1 | my_str_col
  14. ----- +-----------
  15.   1    |  [qxl,   电, 力, 设, 备, 杂, 志]
  16.   2    |  [qxl,  电, 力, 设, 备,  杂, 志]
  18. # 实现
  19. # 注意jieba分词的文件路径和导入方式
  20. import re
  21. import jieba
  23. from pyspark.sql.functions import UserDefinedFunction
  24. from pyspark.sql.types import ArrayType, StringType
  25. from pyspark.sql.functions import udf
  28. # 实现期待结果 1
  29. @udf(returnType=ArrayType(StringType()))
  30. def splitToken(cols):
  31.     """
  32.         1. 输入参数长度可变，字段可变
  33.         2. 正则去除特殊字符
  34.     """
  35.     col=""
  36.     for i in cols:
  37.         if i is not None:
  38.             col+=i
  39.     col = re.sub('\W+', ' ', col).replace("_", '').replace(" ", '')
  40.     if not jieba.dt.initialized:
  41.         jieba.load_userdict("/opt/cloudera/parcels/Anaconda-5.2.0/envs/py3.6/lib/python3.6/site-packages/jieba/dict.txt")
  42.     return [i for i in col]    #  视情况修改返回值格式
  44. # 实现期待结果 2
  45. def splitEN_CZ(oneStr):
  46.     """分隔中英文字符串"""
  47.     uncn = re.compile(r'[^\u4e00-\u9fa5]')
  48.     en = "".join(uncn.findall(oneStr))
  49.     en = re.sub('\W+', ' ', en).replace("_", '')
  50.     uncz = re.compile(r'[\u4e00-\u9fa5]')
  51.     cz = "".join(uncz.findall(oneStr))
  52.     cz = re.sub('\W+', ' ', cz).replace("_", '')
  53.     tmp =[j for j in cz] + [j for j in en.split()]
  54.     return tmp
  56. @udf(returnType=ArrayType(StringType()))
  57. def splitToken(cols):
  58.     """
  59.         my_func()函数的改进
  60.         1. 输入参数长度可变，字段可变
  61.         2. 正则去除特殊字符
  62.     """
  63.     col=""
  64.     for i in cols:
  65.         if i is not None:
  66.             col+= " ".join(splitEN_CZ(i))
  67.     col = re.sub('\W+', ' ', col).replace("_", '')
  68.     if not jieba.dt.initialized:
  69.         jieba.load_userdict("/opt/cloudera/parcels/Anaconda-5.2.0/envs/py3.6/lib/python3.6/site-packages/jieba/dict.txt")
  70.     return col.split() #  视情况修改返回值格式
  72. df = df.withColumn('my_array_col', splitToken(array('my_str_col')) )
  73. # 同时对多列操作
  74. df = df.withColumn('my_array_col',splitToken(array(col("my_str_col1"), col("my_str_col2"))))

3. 将一列string转为多列string

更多例子查看https://stackoverflow.com/questions/39235704/split-spark-dataframe-string-column-into-multiple-columns
使用 函数 pyspark.sql.functions.split()

  1. # 原DataFrame
  2. col1 | my_str_col
  3. -----+-----------
  4.   18 |  856-yygrm
  5.  201 |  777-psgdg
  7. # 期待结果
  8. col1 | my_str_col | _col3 | _col4
  9. -----+------------+-------+------
  10.   18 |  856-yygrm |   856 | yygrm
  11.  201 |  777-psgdg |   777 | psgdg
  13. # 操作
  14. split_col = pyspark.sql.functions.split(df['my_str_col'], '-')
  15. df = df.withColumn('NAME1', split_col.getItem(0))
  16. df = df.withColumn('NAME2', split_col.getItem(1))
  18. #结果
  19. col1 | my_str_col | NAME1 | NAME2
  20. -----+------------+-------+------
  21.   18 |  856-yygrm |   856 | yygrm
  22.  201 |  777-psgdg |   777 | psgdg

4. 对两个df实现行的拼接(第2列接到第1列末尾)

  1. df1=df.select(df['age'],df['name'])
  2. df2=df.select(df['age'],df['weigh'])
  3. df1.unionAll(df2).show()

5. 增加一列有序id，从0计数

from pyspark.sql.functions import monotonically_increasing_id
df.withColumn(‘id’,monotonically_increasing_id() )

6. 将一个array()扩展成一列

用 explode()

  1. # 数据
  2. |id|  label   |
  3. |0  |['a','b']|
  5. # 期待
  6. |id|  label   |  after|
  7. |0  |['a','b']|      a   |
  8. |0  |['a','b']|      b   |
  10. df.withColumn('after', explode(col('label'))).show()

7. 将一个字符串拆分成每行一个

结合使用explode、split

  1. .withColumn('one_item', explode(split('items', ',')))