scala - 将 Spark 数据帧保存到按日期分区的 HDFS
问题描述
我需要将 Spark 数据帧中的数据以 Avro 格式写入 HDFS。挑战在于数据应该每天保存,所以目录看起来像这样:tablename/2019-08-12、tablename/2019-08-13 等等。我只有一个时间戳字段,我需要从中提取创建目录名称的日期。我已经建立了一种方法,它有两个问题:1)从时间戳中提取日期有困难 3)在大型数据集(以后会更大)上,随着许多任务的启动,性能将非常糟糕。那么我该如何改变/改进这种方法呢?
这是我使用的代码(dataDF 是输入数据):
val uniqueDates = dataDF.select("update_database_time").distinct.
collect.map(elem => elem.getTimestamp(0).getDate)
uniqueDates.map(date => {
val resultDF = dataDF.where(to_date(dataDF.col("update_database_time")) <=> date)
val pathToSave = s"${dataDir}/${tableNameValue}/${date}"
dataDF.write
.format("avro")
.option("avroSchema", SchemaRegistry.getSchema(
schemaRegistryConfig.url,
schemaRegistryConfig.dataSchemaSubject,
schemaRegistryConfig.dataSchemaVersion))
.save(s"${hdfsURL}${pathToSave}")
resultDF
})
.reduce(_.union(_))
解决方案
如果您可以使用目录结构,例如
tablename/date=2019-08-12
tablename/date=2019-08-13
相反,然后DataFrameWriter.partitionBy
就可以了。例如
val df =
Seq((Timestamp.valueOf("2019-06-01 12:00:00"), 1),
(Timestamp.valueOf("2019-06-01 12:00:01"), 2),
(Timestamp.valueOf("2019-06-02 12:00:00"), 3)).toDF("time", "foo")
df.withColumn("date", to_date($"time"))
.write
.partitionBy("date")
.format("avro")
.save("/tmp/foo")
产生以下结构
find /tmp/foo
/tmp/foo
/tmp/foo/._SUCCESS.crc
/tmp/foo/date=2019-06-01
/tmp/foo/date=2019-06-01/.part-00000-2a7a63f2-7038-4aec-8f76-87077f91a415.c000.avro.crc
/tmp/foo/date=2019-06-01/part-00000-2a7a63f2-7038-4aec-8f76-87077f91a415.c000.avro
/tmp/foo/date=2019-06-01/.part-00001-2a7a63f2-7038-4aec-8f76-87077f91a415.c000.avro.crc
/tmp/foo/date=2019-06-01/part-00001-2a7a63f2-7038-4aec-8f76-87077f91a415.c000.avro
/tmp/foo/_SUCCESS
/tmp/foo/date=2019-06-02
/tmp/foo/date=2019-06-02/part-00002-2a7a63f2-7038-4aec-8f76-87077f91a415.c000.avro
/tmp/foo/date=2019-06-02/.part-00002-2a7a63f2-7038-4aec-8f76-87077f91a415.c000.avro.crc
推荐阅读
- java - 如何在窗格上添加多个灯光效果?
- python - 如何使用 selenium 与 dcc.DatePickerSingle 交互
- python - Yaml 配置文件包含 webdriver.Firefox() 方法作为参数,如何转换字符串以便将其分配给驱动程序函数
- mysql - 试图找到单词 down and dome (LIKE)
- angular - Activationstart 上的单元测试 Angular 路由器事件
- javascript - 如何将此数据映射到数据属性?
- javascript - 使用Ajax,laravel在Modal中显示错误消息
- django - 如何防止 Django 模型中的 IntegrityError
- python - 在windows RDP中无法使用python exe中的chromedriver连接到服务
- arrays - 分析特定方向的阵列