0

Estoy intentando hacer un upsert endeltalake

val updatesDF = ...  // define the updates DataFrame[date, eventId, data]

DeltaTable.forPath(spark, "/data/events/")
  .as("events")
  .merge(
    updatesDF.as("updates"),
    "events.eventId = updates.eventId")
  .whenMatched
  .updateExpr(
    Map("data" -> "updates.data"))
  .whenNotMatched
  .insertExpr(
    Map(
      "date" -> "updates.date",
      "eventId" -> "updates.eventId",
      "data" -> "updates.data"))
  .execute()

Pero este escribe muchos archivos .parquet (1 por cada fila!) ¿que puede estar pasando?

Gracias!!

1
  • Bienvenido al sitio, traduce tu pregunta, el idioma oficial del sitio es el "español", saludos.
    – ivanao
    el 1 abr. 2020 a las 7:58

1 respuesta 1

0

Spark si tiene algún tipo de shuffle (groupBy, join etc) reparte tus datos en particiones. Por defecto spark pone 200 particiones, pero se puede configurar con la opción spark.sql.shuffle.partitions. Estas particiones son el número de tareas que divide el job (trabajo) en un stage o etapa, y si este stage es el último, escribirá 200 ficheros.

Reduce este valor y escribirá un menor número de ficheros.

Tu Respuesta

By clicking “Publica tu respuesta”, you agree to our terms of service and acknowledge you have read our privacy policy.

¿No es la respuesta que buscas? Examina otras preguntas con la etiqueta o formula tu propia pregunta.