处理pyspark/hive中不断变化的数据类型

31moq8wy 于 2021-05-24 发布在 Spark

关注(0)|答案(1)|浏览(685)

我在分析pyspark中不一致的数据类型时遇到了一个问题。如下面的示例文件所示，sa key始终包含一个字典，但有时它可以显示为字符串值。当我尝试获取列sa.sm.name时，得到如下所示的异常。
如何在pyspark/hive中将sa.sm.name列的值设为null，以表示除json以外的值。有人能帮我吗？
我试图转换到不同的数据类型，但没有任何工作，或可能是我会做一些错误的事情。
输入文件内容：mypath
{“id”：1，“sa”：{“sm”：{“name”：“john”，“email”：john@example.com"}}}
{“id”：2，“sa”：{“sm”：{“name”：“jerry”，“email”：jerry@example.com"}}}
{“id”：3，“sa”：“stringvalue”}

df=spark.read.json(my_path)
df.registerTempTable("T")
spark.sql("""select id,SA.SM.Name from T """).show()

traceback（最近一次调用last）：文件“，”第1行，在文件“/usr/lib/spark/python/pyspark/sql/session.py”中，第767行，在sql return dataframe（self.\u jsparksession.sql（sqlquery），self.\u wrapped）文件“/usr/lib/spark/python/lib/py4j-0.10.7-src.zip/py4j/java\u gateway”中，第1257行，在调用文件“/usr/lib/spark/pyspark/sql/utils.py”中，第69行，在deco raise analysisexception（s.split（'：'，1）[1]，stacktrace）pyspark.sql.utils.analysisexception中：“无法从sa#6.sm提取值：需要struct type，但得到string；1号线位置10“

python apache-spark pyspark apache-spark-sql

来源：https://stackoverflow.com/questions/64289718/handling-changing-datatypes-in-pyspark-hive

1条答案

按热度按时间

bgibtngc1#

使用dataframes是不可能的，因为spark加载sa列时，它被读取为字符串。但是可以使用sparkcontext作为rdd加载文件/表，然后使用cleaner函数将空dict值Map到sa。在这里，我将文件加载为textfile，但如果它是hadoopfile，则执行必要的实现。

def cleaner(record):
    output = ""
    print(type(record))
    try:
        output = json.loads(record)
    except Exception as e:
        print("exception happened")
    finally:
        if isinstance(output.get("SA"), str ):
            print("This is string")
            output["SA"] = {}
    return output

dfx = spark.sparkContext.textFile("file://"+my_path)

dfx2 = dfx.map(cleaner)

new_df = spark.createDataFrame(dfx2)
new_df.show(truncate=False)
+---------------------------------------------------+---+
|SA                                                 |id |
+---------------------------------------------------+---+
|[SM -> [Email -> John@example.com, Name -> John]]  |1  |
|[SM -> [Email -> Jerry@example.com, Name -> Jerry]]|2  |
|[]                                                 |3  |
+---------------------------------------------------+---+

new_df.printSchema()
root
 |-- SA: map (nullable = true)
 |    |-- key: string
 |    |-- value: map (valueContainsNull = true)
 |    |    |-- key: string
 |    |    |-- value: string (valueContainsNull = true)
 |-- id: long (nullable = true)

注意：如果name的输出值必须写入同一个表/列，此解决方案可能不起作用，并且如果您尝试将加载的Dataframe写回同一个表，那么它将导致sa列中断，并且您将根据qn的注解中提供的架构得到一个名称和电子邮件的列表。

赞(0）回复(0）举报 2021-05-25

我来回答

处理pyspark/hive中不断变化的数据类型

1条答案

相关问题

热门标签

最新问答