hadoop - 在非结构化文件上进行 Spark 提取和转换

Question

我的文件夹在本地目录中包含许多销售文本文件。让我们以2个文本文件为例:

文字档1:

Sales Details
20161120


Sales Person: John

Code    Product     Quantity    Price
A0001   Product1    20          15.90
A0003   Product3    13          23.80

文字档2:

Sales Details
20161130


Sales Person: Alicia

Code    Product     Quantity    Price
A0007   Product7    342         79.50
A0008   Product8    55          432.80
A0009   Product9    100         134.30

我曾经用Flume将文件流式传输到HDFS中。所有这些小文件在HDFS中被合并为一个大文件。当我使用Spark对这些文件执行提取和转换时，遇到了一些问题，需要在这里向大家寻求建议。

基于以上两个文件，它将在HDFS中合并为一个文件。我使用Spark从HDFS读取文本文件，如下所示:

lines = spark.read.text('/user/tester/sales')

如何将 行分成两个销售明细，然后为每个销售人员提取信息？我的最终目标是使用以下结构提取信息并将其放入Hive表中:

Date     SalesPerson     Code     Product     Quantity    Price

谢谢。

Answer 1

您的文件结构不是很方便处理，但是您始终可以将带有Spark的wholeTextFiles的正则表达式改写为表格格式。请参阅以下pyspark代码作为示例:

import re

def extract_sales(file):
    for line in file[1].split("\n"):
        if re.match('\d{8}', line.strip()):
            date = line.strip()
        if re.search('^Sales Person', line):
            person = re.match("^Sales Person: (.*)", line).group(1)
        if re.search('^A00', line):
            yield [date, person] + re.split('\s+', line)

raw_data = spark.sparkContext.wholeTextFiles('sales/')
raw_data.flatMap(extract_sales) \
    .toDF(['Date', 'SalesPerson', 'Code', 'Product', 'Quantity', 'Price']).show()

+--------+-----------+-----+--------+--------+------+
|    Date|SalesPerson| Code| Product|Quantity| Price|
+--------+-----------+-----+--------+--------+------+
|20161120|       John|A0001|Product1|      20| 15.90|
|20161120|       John|A0003|Product3|      13| 23.80|
|20161130|     Alicia|A0007|Product7|     342| 79.50|
|20161130|     Alicia|A0008|Product8|      55|432.80|
|20161130|     Alicia|A0009|Product9|     100|134.30|
+--------+-----------+-----+--------+--------+------+