Estoy trabajando con AWS y tengo flujos de trabajo que usan Spark y Hive. Mis datos están particionados por fecha, por lo que todos los días tengo una nueva partición en mi almacenamiento S3. Mi problema es cuando un día falla la carga de datos y tengo que volver a ejecutar esa partición. El código que escribe es el siguiente:
df // My data in a Dataframe .write .format(getFormat(target)) // csv by default, but could be parquet, ORC... .mode(getSaveMode("overwrite")) // Append by default, but in future it should be Overwrite .partitionBy(partitionName) // Column of the partition, the date .options(target.options) // header, separator... .option("path", target.path) // the path where it will be storage .saveAsTable(target.tableName) // the table name¿Qué sucede en mi flujo? Si uso SaveMode.Overwrite, la tabla completa se eliminará y solo tendré guardada la partición. Si uso SaveMode.Append, podría tener datos duplicados.
Haciendo una búsqueda, encontré que Hive soporta este tipo de sobreescritura, solo partición, pero usando las sentencias hql, no lo tengo.
Necesitamos la solución en Hive, por lo que no podemos usar esta opción alternativa (directo a csv).
Había encontrado este ticket de Jira que se supone que soluciona el problema que tengo, pero al intentarlo con la última versión de Spark (2.3.0), la situación era la misma. Elimina toda la tabla y guarda la partición en lugar de sobrescribir la partición que tienen mis datos.
Tratando de aclarar esto, este es un ejemplo:
Particionado por A
Datos:
| A | B | C | |---|---|---| | b | 1 | 2 | | c | 1 | 2 |Mesa:
| A | B | C | |---|---|---| | a | 1 | 2 | | b | 5 | 2 | Lo que quiero es: en la tabla, la partición a permanece en la tabla, la partición b se sobrescribe con los datos y agrega la partición c . ¿Hay alguna solución usando Spark que pueda hacer esto?
Mi última opción para hacer esto es primero eliminar la partición que se va a guardar y luego usar SaveMode.Append, pero intentaría esto en caso de que no haya otra solución.
Si está en Spark 2.3.0, intente establecer la configuración spark.sql.sources.partitionOverwriteMode en dynamic , el conjunto de datos debe particionarse y el modo de escritura sobrescribir.
spark.conf.set("spark.sql.sources.partitionOverwriteMode","dynamic") data.write.mode("overwrite").insertInto("partitioned_table")Por lo tanto, si está utilizando la versión de Spark < 2.3 y desea escribir en particiones de forma dinámica sin eliminar las demás, puede implementar la siguiente solución.
La idea es registrar el conjunto de datos como una tabla y luego usar spark.sql() para ejecutar la consulta INSERT.
// Create SparkSession with Hive dynamic partitioning enabled val spark: SparkSession = SparkSession .builder() .appName("StatsAnalyzer") .enableHiveSupport() .config("hive.exec.dynamic.partition", "true") .config("hive.exec.dynamic.partition.mode", "nonstrict") .getOrCreate() // Register the dataframe as a Hive table impressionsDF.createOrReplaceTempView("impressions_dataframe") // Create the output Hive table spark.sql( s""" |CREATE EXTERNAL TABLE stats ( | ad STRING, | impressions INT, | clicks INT |) PARTITIONED BY (country STRING, year INT, month INT, day INT) |ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LINES TERMINATED BY '\n' """.stripMargin ) // Write the data into disk as Hive partitions spark.sql( s""" |INSERT OVERWRITE TABLE stats |PARTITION(country = 'US', year = 2017, month = 3, day) |SELECT ad, SUM(impressions), SUM(clicks), day |FROM impressions_dataframe |GROUP BY ad """.stripMargin )Sugeriría ejecutar sql usando sparksession. puede ejecutar "insertar sobrescribir consulta de partición" seleccionando las columnas del conjunto de datos existente. esta solución seguramente sobrescribirá la partición solamente.