как разобрать CLOB в pyspark?

Я взял данные из Oracle, и в таблице был столбец с типом данных CLOB, я сделал его строковым, чтобы получить данные в HDFS. Теперь мне нужно разобрать данные CLOB и создать для них отдельную таблицу в Hive.

У меня есть файл HDFS в формате txt. Я могу разделить данные CLOB и надеюсь сделать DataFrame для CLOB.

CLOB имеет следующий формат:

[name] Bob [Age] 21 [City] London [work] No,
[name] Steve [Age] 51 [City] London [work] Yes,
.....
around a million rows like this

sc.setLogLevel("WARN")
log_txt=sc.textFile("/path/to/data/sample_data.txt")
header = log_txt.first()

log_txt = log_txt.filter(lambda line: line != header)
log_txt.take(10)
  [u'0\\tdog\\t20160906182001\\tgoogle.com', u'1\\tcat\\t20151231120504\\tgoogle.com']

temp_var = log_txt.map(lambda k: k.split("\\t"))

log_df=temp_var.toDF(header.split("\\t"))

log_df = log_df.withColumn("field1Int", log_df["field1"].cast(IntegerType()))
log_df = log_df.withColumn("field3TimeStamp", log_df["field1"].cast(TimestampType()))

log_df.schema
StructType(List(StructField(field1,StringType,true),StructField(field2,StringType,true),StructField(field3,StringType,true),StructField(field4,StringType,true),StructField(field1Int,IntegerType,true),StructField(field3TimeStamp,TimestampType,true)))

Вот как я создал DataFrame.

Мне нужна ваша помощь, чтобы понять, как разобрать CLOB, имеющий форму строкового типа данных. и Создайте таблицу поверх этого.

После разборки я ожидаю, что в таблице будут следующие столбцы, например:

+---------+---------------+----------+-----+
|Name     |Age            | City     | Work|
+---------+---------------+----------+-----+
|      Bob|           21  |London    | No  |
|    Steve|           51  |London    |Yes  |
+---------+---------------+----------+-----+

Любая помощь будет оценена по достоинству.


apache-spark pyspark apache-spark-sql pyspark-dataframes pyspark-sql
person sareen    schedule 12.11.2019    source источник

Ответы (1)


arrow_upward
0
arrow_downward

вот:

import re
from pyspark.sql import Row

rdd = sc.parallelize(["[name] Bob [Age] 21 [City] London [work] No",
                      "[name] Steve [Age] 51 [City] London [work] Yes",
                      "[name] Steve [Age] [City] London [work] Yes"])

def clob_to_table(line):
    m = re.search(r"\[name\](.*)?\[Age\](.*)?\[City\](.*)?\[work\](.*)?", line)
    return Row(name=m.group(1), age=m.group(2), city=m.group(3), work=m.group(4))

rdd = rdd.map(clob_to_table)

df = spark.createDataFrame(rdd)
df.show()

+----+--------+-------+----+
| age|    city|   name|work|
+----+--------+-------+----+
| 21 | London |   Bob |  No|
| 51 | London | Steve | Yes|
|    | London | Steve | Yes
+----+--------+-------+----+
person aleaugustoplus    schedule 13.11.2019
comment
Большое спасибо! Мне нужно было только это. Но есть пара сомнений. Во-первых, я вижу, что вы использовали кавычки в начале и в конце данных, но у меня есть файл с миллионами записей, так что как бы вы порекомендовали это сделать? Во-вторых, что, если в каком-то случае [Город] отсутствует, я попытался, но получил ошибку, я думаю пометить его как NULL всякий раз, когда они не имеют значений. Любые предложения для этого тоже? Спасибо еще раз! - person sareen; 14.11.2019
comment
1 - Этот синтаксис предназначен для загрузки вашего примера без необходимости создания файла, если бы у меня был файл, файл можно было бы загрузить, выполнив следующие действия: text_file = sc.textFile(hdfs://...) Предполагается, что ваш файл сохраните каждую строку в одной строке, если это не тот формат, вам, вероятно, следует выполнить предварительную обработку, которая разбивает файл по одному в строке на строку 2. Это зависит от того, как включены пропущенные экземпляры, вам может потребоваться изменить регулярное выражение, например, если город отсутствует, но метка города включена, это нормально, включите NULL, но если метка «город» отсутствует, измените регулярное выражение - person aleaugustoplus; 14.11.2019