kedro.extras.datasets.spark.SparkHiveDataSet

SparkHiveDataSet loads and saves Spark dataframes stored on Hive. This data set also handles some incompatible file types such as using partitioned parquet on hive which will not normally allow upserts to existing data without a complete replacement of the existing file/partition.

This DataSet has some key assumptions:

• Schemas do not change during the pipeline run (defined PKs must be present for the duration of the pipeline)
• Tables are not being externally modified during upserts. The upsert method is NOT ATOMIC

to external changes to the target table while executing. Upsert methodology works by leveraging Spark DataFrame execution plan checkpointing.

Example usage for the YAML API:

hive_dataset:
  type: spark.SparkHiveDataSet
  database: hive_database
  table: table_name
  write_mode: overwrite

Example usage for the Python API:

>>> from pyspark.sql import SparkSession
>>> from pyspark.sql.types import (StructField, StringType,
>>>                                IntegerType, StructType)
>>>
>>> from kedro.extras.datasets.spark import SparkHiveDataSet
>>>
>>> schema = StructType([StructField("name", StringType(), True),
>>>                      StructField("age", IntegerType(), True)])
>>>
>>> data = [('Alex', 31), ('Bob', 12), ('Clarke', 65), ('Dave', 29)]
>>>
>>> spark_df = SparkSession.builder.getOrCreate().createDataFrame(data, schema)
>>>
>>> data_set = SparkHiveDataSet(database="test_database", table="test_table",
>>>                             write_mode="overwrite")
>>> data_set.save(spark_df)
>>> reloaded = data_set.load()
>>>
>>> reloaded.take(4)