adding jar.txt

spark.jars /opt/hudi/hudi-spark-bundle.jar,/opt/hadoop/share/hadoop/hdfs/lib/hadoop-aws-3.2.0.jar,/usr/local/hive/lib/hive-exec.jar
spark.serializer org.apache.spark.serializer.KryoSerializer

CREATE EXTERNAL TABLE hudi_table (
  id STRING,
  name STRING,
  ts TIMESTAMP
)
STORED AS PARQUET
LOCATION 'hdfs://localhost:9000/user/hudi_table';


// Hudi library to work with Hudi tables
libraryDependencies += "org.apache.hudi" %% "hudi-spark-bundle" % "0.9.0"

// Spark core and SQL libraries to work with Spark
libraryDependencies += "org.apache.spark" %% "spark-core" % "3.2.3"
libraryDependencies += "org.apache.spark" %% "spark-sql" % "3.2.3"



import org.apache.spark.sql.SparkSession
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieWriteConfig._
import org.apache.spark.sql.SaveMode

val spark = SparkSession.builder()
  .appName("Hudi Test")
  .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .getOrCreate()

val data = Seq((1, "Alice", "2023-09-17T12:00:00Z"),
               (2, "Bob", "2023-09-17T12:05:00Z"))

val df = spark.createDataFrame(data).toDF("id", "name", "ts")

df.write.format("hudi")
  .option(TABLE_NAME, "hudi_table")
  .option(RECORDKEY_FIELD_OPT_KEY, "id")
  .option(PRECOMBINE_FIELD_OPT_KEY, "ts")
  .mode(SaveMode.Append)
  .save("hdfs://localhost:9000/user/hudi_table")




spark-shell --conf "spark.executor.extraJavaOptions=-Dlog4j.configuration=file:/path/to/log4j.properties"





from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("Hudi Test") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .config("spark.jars", "/opt/hudi/hudi-spark-bundle.jar") \
    .getOrCreate()


data = [(1, "Alice", "2023-09-17T12:00:00Z"), 
        (2, "Bob", "2023-09-17T12:05:00Z")]

df = spark.createDataFrame(data, ["id", "name", "ts"])


df.write.format("hudi") \
    .option("hoodie.table.name", "hudi_table") \
    .option("hoodie.datasource.write.recordkey.field", "id") \
    .option("hoodie.datasource.write.precombine.field", "ts") \
    .mode("append") \
    .save("hdfs://localhost:9000/user/hudi_table")


USE hudi_test;
SELECT * FROM hudi_table;






import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.SaveMode
import org.apache.hudi.DataSourceWriteOptions // Import only DataSourceWriteOptions

object HudiTest {
  def main(args: Array[String]): Unit = {
    
    // Create Spark session
    val spark = SparkSession.builder()
      .appName("Hudi Test")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.jars", "/opt/hudi/hudi-spark-bundle.jar")  // Specify Hudi jar
      .getOrCreate()

    // Sample data
    val data = Seq(
      (1, "Alice", "2023-09-17T12:00:00Z"),
      (2, "Bob", "2023-09-17T12:05:00Z")
    )

    // Create DataFrame
    val df = spark.createDataFrame(data).toDF("id", "name", "ts")

    // Write Data to Hudi table with qualified references to avoid ambiguity
    df.write.format("hudi")
      .option(DataSourceWriteOptions.TABLE_NAME, "hudi_table_test") // Qualified reference
      .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "id") // Qualified reference
      .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "ts") // Qualified reference
      .mode(SaveMode.Append)
      .save("hdfs://localhost:9000/user/hudi_table_test")

    // Stop the Spark session
    spark.stop()
  }
}



import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.SaveMode
import org.apache.hudi.DataSourceWriteOptions

object HudiTest {
  def main(args: Array[String]): Unit = {

    // Create Spark session
    val spark = SparkSession.builder()
      .appName("Hudi Test")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.jars", "/opt/hudi/hudi-spark-bundle.jar")  // Specify Hudi jar
      .getOrCreate()

    // Sample data
    val data = Seq(
      (1, "Alice", "2023-09-17T12:00:00Z"),
      (2, "Bob", "2023-09-17T12:05:00Z")
    )

    // Create DataFrame
    val df = spark.createDataFrame(data).toDF("id", "name", "ts")

    // Write Data to Hudi table using string keys instead of ConfigProperty
    df.write.format("hudi")
      .option("hoodie.table.name", "hudi_table_test")  // Use string key instead of ConfigProperty
      .option("hoodie.datas
ource.write.recordkey.field", "id")  // Use string key
      .option("hoodie.datasource.write.precombine.field", "ts")  // Use string key
      .mode(SaveMode.Append)
      .save("hdfs://localhost:9000/user/hudi_table_test")

    // Stop the Spark session
    spark.stop()
  }
}





Step 5: Integrating Apache Hudi with Spark
5.1 Add Hudi Dependencies to Spark:
Use the following command to start Spark Shell with Hudi support:
$SPARK_HOME/bin/spark-shell \
--packages org.apache.hudi:hudi-spark3.3-bundle_2.12:0.13.0 \
--conf "spark.serializer=org.apache.spark.serializer.KryoSerializer" \
--conf "spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension" \
--conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog"

5.2 Running a Sample Hudi Job in Spark:

Write Data to Hudi:
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieWriteConfig._
import org.apache.spark.sql.SaveMode._

val dataGen = spark.range(0, 1000).toDF("id")
dataGen.write.format("hudi")
    .option(TABLE_NAME, "hudi_table")
    .option(PRECOMBINE_FIELD.key(), "id")
    .option(RECORDKEY_FIELD.key(), "id")
    .option(PARTITIONPATH_FIELD.key(), "")
    .option(OPERATION.key(), "insert")
    .option(TABLE_TYPE.key(), "COPY_ON_WRITE")
    .mode(Overwrite)
    .save("/path/to/hudi-table")

Read Hudi Data:
val hudi_df = spark.read.format("hudi").load("/path/to/hudi-table/*")
hudi_df.show()

Step 6: Integrating Apache Hudi with Hive
6.1 Enable Hive Sync for Hudi Tables:
When writing data into a Hudi table using Spark, you can enable Hive sync:

dataGen.write.format("hudi")
    .option(TABLE_NAME, "hudi_table")
    .option(PRECOMBINE_FIELD.key(), "id")
    .option(RECORDKEY_FIELD.key(), "id")
    .option(PARTITIONPATH_FIELD.key(), "")
    .option(OPERATION.key(), "insert")
    .option(TABLE_TYPE.key(), "COPY_ON_WRITE")
    .option("hoodie.datasource.hive_sync.enable", "true")
    .option("hoodie.datasource.hive_sync.database", "default")
    .option("hoodie.datasource.hive_sync.table", "hudi_table")
    .option("hoodie.datasource.hive_sync.jdbcurl", "jdbc:hive2://localhost:10000")
    .option("hoodie.datasource.hive_sync.partition_fields", "date")
    .mode(Overwrite)
    .save("/path/to/hudi-table")

6.2 Query Hudi Table from Hive:
Start the Hive shell and run a query:
hive
SELECT * FROM hudi_table;





import org.apache.hudi.DataSourceWriteOptions // Import only DataSourceWriteOptions
import org.apache.spark.sql.SaveMode

object HudiTest {
  def main(args: Array[String]): Unit = {

    // Create Spark session
    val spark = SparkSession.builder()
      .appName("Hudi Test")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.jars", "/opt/hudi/hudi-spark-bundle.jar")  // Specify Hudi jar
      .getOrCreate()

    // Sample data
    val dataGen = spark.range(0, 1000).toDF("id")

    // Write data to Hudi table using qualified references to avoid ambiguity
    dataGen.write.format("hudi")
      .option(DataSourceWriteOptions.TABLE_NAME.key(), "hudi_table_test")  // Fully qualified reference
      .option(DataSourceWriteOptions.PRECOMBINE_FIELD.key(), "id")         // Fully qualified reference
      .option(DataSourceWriteOptions.RECORDKEY_FIELD.key(), "id")          // Fully qualified reference
      .option(DataSourceWriteOptions.PARTITIONPATH_FIELD.key(), "")        // Fully qualified reference
      .option(DataSourceWriteOptions.OPERATION.key(), "insert")            // Fully qualified reference
      .option(DataSourceWriteOptions.TABLE_TYPE.key(), "COPY_ON_WRITE")    // Fully qualified reference
      .mode(SaveMode.Overwrite)                                            // Overwrite existing data
      .save("hdfs://localhost:9000/user/hudi/hudi_table_test")             // Save to HDFS

    // Stop the Spark session
    spark.stop()
  }
}


import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieWriteConfig._
import org.apache.spark.sql.SaveMode

val tableName = "hudi_test_table"
val basePath = "file:///tmp/hudi_test_table"

val data = Seq(
  (1, "Alice", 24),
  (2, "Bob", 30),
  (3, "Cathy", 27)
)

val df = spark.createDataFrame(data).toDF("id", "name", "age")

df.write.format("hudi")
  .option(TABLE_NAME, tableName)
  .option(RECORDKEY_FIELD_OPT_KEY, "id")
  .option(PRECOMBINE_FIELD_OPT_KEY, "age")
  .mode(SaveMode.Overwrite)
  .save(basePath)




./spark-shell --jars /opt/hudi/hudi-spark3.1.2-bundle_2.12-0.10.1.jar \
              --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
              --conf spark.sql.hive.convertMetastoreParquet=false


CREATE EXTERNAL TABLE hudi_hive_table (
    id INT,
    name STRING,
    age INT
)
STORED BY 'org.apache.hudi.hadoop.HoodieHiveStorageHandler'
LOCATION 'file:///tmp/hudi_test_table'
TBLPROPERTIES (
    'hoodie.datasource.write.recordkey.field'='id',
    'hoodie.datasource.write.precombine.field'='age',
    'hoodie.table.name'='hudi_test_table',
    'hoodie.datasource.write.hive_style_partitioning'='true'
);
export HADOOP_CLASSPATH=/opt/hudi/hudi-spark3.1.2-bundle_2.12-0.10.1.jar:$HADOOP_CLASSPATH


SELECT * FROM hudi_hive_table;


hive> CREATE EXTERNAL TABLE hudi_hive_table(
    >     id INT,
    >     name STRING,
    >     age INT
    > )
    > STORED BY 'org.apache.hudi.hadoop.HoodieParquetInputFormat'
    > LOCATION 'file:///tmp/hudi_test_table';
FAILED: Execution Error, return code 1 from org.apache.hadoop.hive.ql.exec.DDLTask. org.apache.hudi.hadoop.HoodieParquetInputFormat cannot be cast to org.apache.hadoop.hive.ql.metadata.HiveStorageHandler