我在Scala 2.4.0中有一个大数据框架,看起来像
+--------------------+--------------------+--------------------+-------------------+--------------+------+
| cookie| updated_score| probability| date_last_score|partition_date|target|
+--------------------+--------------------+--------------------+-------------------+--------------+------+
|00000000000001074780| 0.1110987111481027| 0.27492987342938174|2019-03-29 16:00:00| 2019-04-07_10| 0|
|00000000000001673799| 0.02621894072693878| 0.2029688362968775|2019-03-19 08:00:00| 2019-04-07_10| 0|
|00000000000002147908| 0.18922034021212567| 0.3520678649755828|2019-03-31 19:00:00| 2019-04-09_12| 1|
|00000000000004028302| 0.06803669083452231| 0.23089047208736854|2019-03-25 17:00:00| 2019-04-07_10| 0|
和此模式:
root
|-- cookie: string (nullable = true)
|-- updated_score: double (nullable = true)
|-- probability: double (nullable = true)
|-- date_last_score: string (nullable = true)
|-- partition_date: string (nullable = true)
|-- target: integer (nullable = false)
然后,我创建一个分区表,然后将数据插入数据库.table_name。但是,当我查找Hive数据库和类型:显示分区数据库。
我不知道我写的是错误的,有一些我不理解的scala概念或数据框太大。
我尝试了不同的方法来查找类似的问题:
result_df.write.mode(SaveMode.Overwrite).insertInto("table_name")
或
result_df.write.mode(SaveMode.Overwrite).saveAsTable("table_name")
如果它有助于我提供Scala的一些信息消息:
查看此消息,我认为我正确地得到了我的result_df分区。
19/07/31 07:53:57 INFO TaskSetManager: Starting task 11.0 in stage 2822.0 (TID 123456, ip-xx-xx-xx.aws.local.somewhere, executor 45, partition 11, PROCESS_LOCAL, 7767 bytes)
19/07/31 07:53:57 INFO TaskSetManager: Starting task 61.0 in stage 2815.0 (TID 123457, ip-xx-xx-xx-xyz.aws.local.somewhere, executor 33, partition 61, NODE_LOCAL, 8095 bytes)
然后,我开始将分区保存为向量(0、1、2 ...(,但是我只能保存0和1?我真的不知道。
19/07/31 07:56:02 INFO DAGScheduler: Submitting 35 missing tasks from ShuffleMapStage 2967 (MapPartitionsRDD[130590] at insertInto at evaluate_decay_factor.scala:165) (first 15 tasks are for partitions Vector(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14))
19/07/31 07:56:02 INFO YarnScheduler: Adding task set 2967.0 with 35 tasks
19/07/31 07:56:02 INFO DAGScheduler: Submitting ShuffleMapStage 2965 (MapPartitionsRDD[130578] at insertInto at evaluate_decay_factor.scala:165), which has no missing parents
我的代码看起来像这样:
val createTableSQL = s"""
CREATE TABLE IF NOT EXISTS table_name (
cookie string,
updated_score float,
probability float,
date_last_score string,
target int
)
PARTITIONED BY (partition_date string)
STORED AS PARQUET
TBLPROPERTIES ('PARQUET.COMPRESSION'='SNAPPY')
"""
spark.sql(createTableSQL)
result_df.write.mode(SaveMode.Overwrite).insertInto("table_name")
给出这样的数据框:
val result = Seq(
(8, "123", 1.2, 0.5, "bat", "2019-04-04_9"),
(64, "451", 3.2, -0.5, "mouse", "2019-04-04_12"),
(-27, "613", 8.2, 1.5, "horse", "2019-04-04_10"),
(-37, "513", 4.33, 2.5, "horse", "2019-04-04_11"),
(45, "516", -3.3, 3.4, "bat", "2019-04-04_10"),
(12, "781", 1.2, 5.5, "horse", "2019-04-04_11")
我想运行:在Hive命令行上显示" table_name"并获取:
partition_date=2019-04-04_9
partition_date=2019-04-04_10
partition_date=2019-04-04_11
partition_date=2019-04-04_12
我的输出中是:
partition_date=0
partition_date=1
在这个简单的示例情况下,它可以很好地工作,但是在我的大数据框架中,我获得了先前的输出。
要更改分区数,使用repartition(numOfPartitions)
要通过编写时通过划分的列来更改列,请使用partitionBy("col")
一起使用的示例:final_df.repartition(40).write.partitionBy("txnDate").mode("append").parquet(destination)
两个有用的提示:
- 使您的重新分配大小等于最快的写入/重新分配的工作核心数量。在此示例中,我有10位执行人,每个执行者有4个内核(总计40个核心(。因此,我将其设置为40。
- 当您写入目的地时,不要指定除子字符以外的任何内容 - 让Spark处理索引。
- 好的目的地:
"s3a://prod/subbucket/"
- 不良目的地:
s"s3a://prod/subbucket/txndate=$txndate"
- 好的目的地: