嗨,我有 90 GB 的数据 在 CSV 文件中,我正在使用选择插入命令将此数据加载到一个临时表中,然后从临时表加载到 orc 表中,但为了将数据转换和加载为 orc 格式,在 Spark SQL 中需要 4 小时。我可以使用任何一种优化技术来减少这段时间。到目前为止,我没有使用任何类型的优化技术,我只是使用 spark sql 并将数据从 csv 文件加载到表(文本格式),然后从这个临时表加载到 orc 表(使用选择插入) 使用Spark提交为:
spark-submit
--class class-name
--jar file
或者我可以在火花提交中添加任何额外的参数以改进优化。
斯卡拉代码(示例):
All Imports
object demo {
def main(args: Array[String]) {
//sparksession with enabled hivesuppport
var a1=sparksession.sql("load data inpath 'filepath' overwrite into table table_name")
var b1=sparksession.sql("insert into tablename (all_column) select 'ALL_COLUMNS' from source_table")
}
}
我只是使用 spark sql 并将数据从 csv 文件加载到 表(文本格式),然后从这个临时表到ORC表(使用 选择插入)
这里不需要 2 步过程。.
- 读取数据帧,如以下示例所示...
val DFCsv = spark.read.format("csv")
.option("sep", ",")
.option("inferSchema", "true")
.option("header", "true")
.load("yourcsv")
- 如果需要,您必须执行
repartition
(这可能是实际 4 小时延迟的原因,因为您还没有这样做),因为它的大文件,然后......
dfcsv.repartition(90)
意味着它将/可能会将 CSV 数据重新划分为 90 个几乎相等的部分。 其中 90 是样本编号。 你可以提到你想要的。
DFCsv.write.format("orc")
.partitionBy('yourpartitioncolumns')
.saveAsTable('yourtable')
或
DFCsv.write.format("orc")
.partitionBy('yourpartitioncolumns')
.insertInto('yourtable')
注意:1)对于大数据,您需要进行重新分区以均匀分布数据,这将增加Parllelism,因此 性能。
2)如果您没有patition列并且 非分区表,则无需在上面
partitionBy
样品