Écritures et streaming

HWC prend en charge les écritures par lots et les écritures Structured Streaming vers les tables Hive ACID.

Écritures par lots (DataFrame writer)

HWC écrit les fichiers de staging sur HDFS puis exécute LOAD DATA vers la table cible. Le mode de lecture n'affecte pas le comportement des écritures.

spark.range(0, 100)
  .selectExpr("id", "concat('v', id) as v")
  .write
  .format("com.hortonworks.spark.sql.hive.llap.HiveWarehouseConnector")
  .option("database", "hwc_it")
  .option("table", "t_acid")
  .mode("overwrite")
  .save()

Remarques :

• Créer la table cible au préalable (Spark 3 ne crée pas automatiquement les tables Hive à l'écriture).
• Utiliser un répertoire de staging qualifié pour l'accès sécurisé et les écritures.

Écritures en streaming (Structured Streaming)

Utiliser le sink de streaming pour écrire vers les tables ACID :

import org.apache.spark.sql.streaming.Trigger

val q = spark.readStream
  .format("rate")
  .option("rowsPerSecond", 5)
  .load()
  .selectExpr("cast(timestamp as string) as ts", "value")
  .writeStream
  .format("com.hortonworks.spark.sql.hive.llap.streaming.HiveStreamingDataSource")
  .outputMode("append")
  .option("database", "hwc_it")
  .option("table", "t_stream")
  .option("metastoreUri", "thrift://hms-host:9083")
  .option("checkpointLocation", "hdfs://nameservice/tmp/hwc_ckpt")
  .trigger(Trigger.Once())
  .start()

Remarques sur le streaming :

• La table cible doit être transactionnelle (ACID).
• metastoreUri est requis pour le streaming.
• Utiliser cleanUpStreamingMeta pour supprimer les métadonnées d'une requête arrêtée.