Фильтрация данных текстового файла в виде столбцов в pyspark rdd и dataframe

Question

У меня есть данные, подобные приведенным ниже:

It has survived not only five centuries, but also the leap into electronic typesetting, remaining essentially unchanged. 
It was popularised in the 1960s with the release of Image sheets containing Buddy passages, and more recently with desktop publishing
 software like

1   long title 1
2 long title 2
3 long title 3
4 long title 4
5 long title 5
6 long title 6
7 long title 7
8 long title 8
9 long title 9
10 long title 10
11 long title 11
12 long title 12
13 long title 13
14 long title 14
15 long title 15
16 long title 16
17 long title 17
18 long title 18
19 long title 19
20 long title 20  

Теперь, при загрузке этого текстового файла, я должен исключить ненужные данные, то есть абзац, и включить данные, начиная с long title 1, то есть столбчатые данные. Я использую RDD, но не могу загрузить его правильно. Как только данные в RDD заполнены правильно, я могу преобразовать их в dataframe. Ниже приведен мой код:

from pyspark.context import SparkContext
from pyspark.sql import SparkSession
from pyspark import SparkConf


sc = SparkContext.getOrCreate(SparkConf().setMaster("local[*]"))
load_data=sc.textFile("E://long_sample.txt").filter(lambda x : "title")
load_data.foreach(print())  

Несмотря на то, что я пытаюсь отфильтровать его по "title", я все еще получаю все данные, которые не верны. Пожалуйста, помогите мне разобраться. Ошибка не отображается.

Answer 1

Вот еще один способ использования rlike и регулярных выражений через API Dataframe:

import pyspark.sql.functions as f
from pyspark.sql.types import StringType

df = spark.createDataFrame([
  ('It was popularised in the 1960s with the release of Image sheets containing Buddy passages, and more recently with desktop publishing'),
  ('software like'),
  ('1   long title 1'),
  ('2 long title 2'),
  ('3 long title 3'),
  ('4 long title 4'),
  ('5 long title 5'),
  ('6 long title 6'),
  ('7 long title 7'),
  ('8 long title 8'),
  ('9 long title 9'),
  ('10 long title 10'),
  ('11 long title 11'),
  ('12 long title 12')
], StringType())

df.where(f.col("value").rlike("\d+\s+\w+\s+\w+\s+\d+")).show(100, False)

# +----------------+
# |           value|
# +----------------+
# |1   long title 1|
# |  2 long title 2|
# |  3 long title 3|
# |  4 long title 4|
# |  5 long title 5|
# |  6 long title 6|
# |  7 long title 7|
# |  8 long title 8|
# |  9 long title 9|
# |10 long title 10|
# |11 long title 11|
# |12 long title 12|
# +----------------+

rlike здесь будет идентифицировать строки, для которых есть соответствие с регулярным выражением \d+\s+\w+\s+\w+\s+\d+. Вот объяснение регулярному выражению:

• \ d +: одна или несколько цифр
• \ s +: после одного или нескольких пробелов
• \ w +: с последующимодна или несколько строчных букв
• \ s +: за которыми следуют один или несколько пробелов
• .....

Если вы уверены, что слова длинные изаголовок всегда присутствует, вы можете изменить регулярное выражение на: \d+\s+long\s+title\s+\d+.

ОБНОВЛЕНИЕ:

Чтобы разбить ваш набор данных на новый набор данных со столбцами id, name, numberиспользуйте выбор и разделите следующим образом:

df.where(df["value"].rlike("\d+\s+long\s+title\s+\d+")) \
  .select(
          f.split(df["value"], "\s+").getItem(0).alias("id"),
          f.concat(f.split(df["value"], "\s+").getItem(1), f.split(df["value"], "\s+").getItem(2)).alias("name"),
          f.split(df["value"], "\s+").getItem(3).alias("number")
  ).show()

# +---+---------+------+
# | id|     name|number|
# +---+---------+------+
# |  1|longtitle|     1|
# |  2|longtitle|     2|
# |  3|longtitle|     3|
# |  4|longtitle|     4|
# |  5|longtitle|     5|
# |  6|longtitle|     6|
# |  7|longtitle|     7|
# |  8|longtitle|     8|
# |  9|longtitle|     9|
# | 10|longtitle|    10|
# | 11|longtitle|    11|
# | 12|longtitle|    12|
# +---+---------+------+

Answer 2

попробуйте это в Pyspark:

>>> load_data=sc.textFile("file:///home/mahesh/Downloads/line_text.txt")

отфильтруйте данные с помощью оператора IN и для создания фрейма данных из существующего RDD

>>> load_data.filter(lambda x: "title" in x).map(lambda x:(x.split(" ")[0],x.split(" ")[1]+" " + x.split(" ")[2],x.split(" ")[3] )).toDF(["Id","Name","Number"])

>>> df.show()
+---+----------+------+
| Id|      Name|Number|
+---+----------+------+
|  1|long title|     1|
|  2|long title|     2|
|  3|long title|     3|
|  4|long title|     4|
|  5|long title|     5|
|  6|long title|     6|
|  7|long title|     7|
|  8|long title|     8|
|  9|long title|     9|
| 10|long title|    10|
| 11|long title|    11|
| 12|long title|    12|
| 13|long title|    13|
| 14|long title|    14|
| 15|long title|    15|
| 16|long title|    16|
| 17|long title|    17|
| 18|long title|    18|
| 19|long title|    19|
| 20|long title|    20|
+---+----------+------+

дайте мне знать, если вынужна дополнительная помощь по тому же.

Answer 3

Пожалуйста, попробуйте также regexp_extract одновременно. Позже вы можете удалить пустые строки.

from pyspark.sql import functions as F

df=spark.read.text('test.txt')

df.select(F.regexp_extract('value', r'(^\d+)\s+(.*)\s+(\d+$)', 1).alias('id')
    ,F.regexp_extract('value',r'(^\d+)\s+(.*)\s+(\d+$)',2).alias('name')
    ,F.regexp_extract('value', r'(^\d+)\s+(.*)\s+(\d+$)', 3).alias('number')
    ).show()