如何有效地将多个json文件读入Dataframe或JavaRDD？

Question

我可以使用以下代码来读取单个json文件,但我需要读取多个json文件并将它们合并到一个Dataframe中.我怎样才能做到这一点？

DataFrame jsondf = sqlContext.read().json("/home/spark/articles/article.json");

或者有没有办法将多个json文件读入JavaRDD然后转换为Dataframe？

Answer 1

要在Spark中读取多个输入,请使用通配符.无论您是构建数据帧还是rdd,情况都是如此.

context.read().json("/home/spark/articles/*.json")
// or getting json out of s3
context.read().json("s3n://bucket/articles/201510*/*.json")

Answer 2

您可以使用完全相同的代码来读取多个JSON文件.只需将路径传递到目录/路径与通配符而不是路径到单个文件.

DataFrameReader还提供json了具有以下签名的方法:

json(jsonRDD: JavaRDD[String])

可用于解析已加载的JSON JavaRDD.

Answer 3

函数spark.read.json接受文件列表作为参数。

spark.read.json(List_all_json file)

这将读取列表中的所有文件并返回文件中所有信息的单个数据框。

Answer 4

使用 pyspark，如果所有 json 文件都在同一文件夹中，则可以使用df = spark.read.json('folder_path'). 该指令将加载文件夹内的所有 json 文件。

为了读取性能，我建议您提供数据帧架构：

import pyspark.sql.types as T

billing_schema = billing_schema = T.StructType([
  T.StructField('accountId', T.LongType(),True),
  T.StructField('accountName',T.StringType(),True),
  T.StructField('accountOwnerEmail',T.StringType(),True),
  T.StructField('additionalInfo',T.StringType(),True),
  T.StructField('chargesBilledSeparately',T.BooleanType(),True),
  T.StructField('consumedQuantity',T.DoubleType(),True),
  T.StructField('consumedService',T.StringType(),True),
  T.StructField('consumedServiceId',T.LongType(),True),
  T.StructField('cost',T.DoubleType(),True),
  T.StructField('costCenter',T.StringType(),True),
  T.StructField('date',T.StringType(),True),
  T.StructField('departmentId',T.LongType(),True),
  T.StructField('departmentName',T.StringType(),True),
  T.StructField('instanceId',T.StringType(),True),
  T.StructField('location',T.StringType(),True),
  T.StructField('meterCategory',T.StringType(),True),
  T.StructField('meterId',T.StringType(),True),
  T.StructField('meterName',T.StringType(),True),
  T.StructField('meterRegion',T.StringType(),True),
  T.StructField('meterSubCategory',T.StringType(),True),
  T.StructField('offerId',T.StringType(),True),
  T.StructField('partNumber',T.StringType(),True),
  T.StructField('product',T.StringType(),True),
  T.StructField('productId',T.LongType(),True),
  T.StructField('resourceGroup',T.StringType(),True),
  T.StructField('resourceGuid',T.StringType(),True),
  T.StructField('resourceLocation',T.StringType(),True),
  T.StructField('resourceLocationId',T.LongType(),True),
  T.StructField('resourceRate',T.DoubleType(),True),
  T.StructField('serviceAdministratorId',T.StringType(),True),
  T.StructField('serviceInfo1',T.StringType(),True),
  T.StructField('serviceInfo2',T.StringType(),True),
  T.StructField('serviceName',T.StringType(),True),
  T.StructField('serviceTier',T.StringType(),True),
  T.StructField('storeServiceIdentifier',T.StringType(),True),
  T.StructField('subscriptionGuid',T.StringType(),True),
  T.StructField('subscriptionId',T.LongType(),True),
  T.StructField('subscriptionName',T.StringType(),True),
  T.StructField('tags',T.StringType(),True),
  T.StructField('unitOfMeasure',T.StringType(),True)
])

billing_df = spark.read.json('/mnt/billingsources/raw-files/202106/', schema=billing_schema)