pyspark解析固定宽度的文本文件答案

【问题标题】：pyspark parse fixed width text filepyspark解析固定宽度的文本文件
【发布时间】：2017-01-30 20:19:11
【问题描述】：

试图解析一个固定宽度的文本文件。

我的文本文件如下所示，我需要一个行 ID、日期、一个字符串和一个整数：

00101292017you1234
00201302017 me5678

我可以使用 sc.textFile(path) 将文本文件读取到 RDD。我可以使用已解析的 RDD 和模式创建 DataFrame。就是这两个步骤之间的解析。

【问题讨论】：

标签： python apache-spark pyspark fixed-width

【解决方案1】：

Spark 的substr 函数可以处理固定宽度的列，例如：

df = spark.read.text("/tmp/sample.txt")
df.select(
    df.value.substr(1,3).alias('id'),
    df.value.substr(4,8).alias('date'),
    df.value.substr(12,3).alias('string'),
    df.value.substr(15,4).cast('integer').alias('integer')
).show()

将导致：

+---+--------+------+-------+
| id|    date|string|integer|
+---+--------+------+-------+
|001|01292017|   you|   1234|
|002|01302017|    me|   5678|
+---+--------+------+-------+

拆分列后，您可以像在普通 spark 数据框中一样重新格式化和使用它们。

【讨论】：

# df = sqlContext.read.text("blah.txt") 我不得不使用 sqlContext。然后它起作用了。我想我需要了解一点上下文。但除此之外，你已经回答了。
spark 代表版本 >= 2.0 中的 spark 会话。如果您使用 spark 1.6 或更低版本，则需要使用 sqlContext，但在访问数据方面的行为几乎相同
有没有办法从字典中获取列名和宽度的值？我想使用字典以编程方式映射我的列。

【解决方案2】：

有人问如何根据架构来做。根据以上回复，这里是一个简单的例子：

x= '''    1 123121234 joe
    2 234234234jill
    3 345345345jane
    4abcde12345jack'''

schema = [
          ("id",1,5),
          ("ssn",6,10),
          ("name",16,4)
]
          
with open("personfixed.csv", "w") as f:
  f.write(x)

df = spark.read.text("personfixed.csv")
df.show()

df2 = df
for colinfo in schema:
  df2 = df2.withColumn(colinfo[0], df2.value.substr(colinfo[1],colinfo[2]))

df2.show()

这是输出：

+-------------------+
|              value|
+-------------------+
|    1 123121234 joe|
|    2 234234234jill|
|    3 345345345jane|
|    4abcde12345jack|
+-------------------+

+-------------------+-----+----------+----+
|              value|   id|       ssn|name|
+-------------------+-----+----------+----+
|    1 123121234 joe|    1| 123121234| joe|
|    2 234234234jill|    2| 234234234|jill|
|    3 345345345jane|    3| 345345345|jane|
|    4abcde12345jack|    4|abcde12345|jack|
+-------------------+-----+----------+----+

【讨论】：

【解决方案3】：

df = spark.read.text("fixedwidth")

df.withColumn("id",df.value.substr(1,5)).withColumn("name",df.value.substr(6,11)).drop('value').show()

结果是

+-----+------+
|   id|  name|
+-----+------+
|23465|ramasg|
|54334|hjsgfd|
|87687|dgftre|
|45365|ghfduh|
+-----+------+

【讨论】：