在创建 Spark DataFrame 时从 hdfs 文件传递​​模式

Question

我正在尝试读取存储在 hdfs 中的文本文件中的架构，并在创建 DataFrame 时使用它。

schema=StructType([
StructField("col1",StringType(),True),
StructField("col2",StringType(),True),
StructField("col3",TimestampType(),True),
StructField("col4",
StructType([
StructField("col5",StringType(),True),
StructField("col6",
.... and so on

jsonDF = spark.read.schema(schema).json('/path/test.json')

由于架构太大，我想在代码中定义。谁能建议最好的方法。

我尝试了以下方法但不起作用。

schema = sc.wholeTextFiles("hdfs://path/sample.schema"))
schema = spark.read.text('/path/sample.schema')

Answer 1

我没有用 hdfs 测试过它，但我认为它类似于从本地文件中读取。这个想法是将文件存储为字典，然后对其进行解析以创建所需的模式。我的灵感来自here。目前它缺乏对 nullable 的支持，我还没有测试过更深层次的嵌套结构。

from pyspark.sql import SparkSession
import pyspark.sql.functions as F
from pyspark.sql.types import *
from fractions import Fraction
from pyspark.sql.functions import udf
import json

spark = SparkSession.builder.appName('myPython').getOrCreate()

f = open("/path/schema_file", "r")

dictString = f.read()

derived_schema = StructType([])

jdata = json.loads(dictString)


def get_type(v):
    if v == "StringType":
        return StringType()
    if v == "TimestampType":
        return TimestampType()
    if v == "IntegerType":
        return IntegerType()


def generate_schema(jdata, derived_schema):
    for k, v in sorted(jdata.items()):
        if (isinstance(v, str)):
            derived_schema.add(StructField(k, get_type(v), True))
        else:
            added_schema = StructType([])
            added_schema = generate_schema(v, added_schema)
            derived_schema.add(StructField(k, added_schema, True))
    return derived_schema


generate_schema(jdata, derived_schema)

from datetime import datetime

data = [("first", "the", datetime.utcnow(), ["as", 1])]

input_df = spark.createDataFrame(data, derived_schema)

input_df.printSchema()

文件是：

{
  "col1" : "StringType",
  "col2" : "StringType",
  "col3" : "TimestampType",
  "col4" : {
    "col5" : "StringType",
    "col6" : "IntegerType"
  }
}

Answer 2

我想出了如何做到这一点。

1. Define the schema of json file 

json.schema=StructType([
StructField("col1",StringType(),True),
StructField("col2",StringType(),True),
StructField("col3",TimestampType(),True),
StructField("col4",
StructType([
StructField("col5",StringType(),True),
StructField("col6",

2. Print the json output

print(sampletmp.json()) 

3. Copy paste the above output to file sample.schema

4. In the code, recreate the schema as below

schema_file = 'path/sample.schema'
schema_json = spark.read.text(schema_file).first()[0]
schema = StructType.fromJson(json.loads(schema_json))

5. Create a DF using above schema

spark.read.schema(schema).json('/path/test.json')

6. Insert the data from DF into Hive table
jsonDF.write.mode("append").insertInto("hivetable")

参考文章-https://szczeles.github.io/Reading-JSON-CSV-and-XML-files-efficiently-in-Apache-Spark/