End to End xtable to Trino query with source delta and target Iceberg and hudi with HMS and S3

[alberttwong]

Using openjdk 11, spark 3.4, hive 2.3.10, hadoop 2.10.2

using https://github.com/alberttwong/incubator-xtable/tree/main/demo-s3 as the base

from pyspark.sql.types import *
from pyspark import SparkContext
from pyspark.sql import SparkSession
import os

# initialize the bucket
database_name = "delta_db"
table_name = "people"
local_base_path = "s3a://warehouse"

records = [
   (1, 'John', 25, 'NYC', '2023-09-28 00:00:00'),
   (2, 'Emily', 30, 'SFO', '2023-09-28 00:00:00'),
   (3, 'Michael', 35, 'ORD', '2023-09-28 00:00:00'),
   (4, 'Andrew', 40, 'NYC', '2023-10-28 00:00:00'),
   (5, 'Bob', 28, 'SEA', '2023-09-23 00:00:00'),
   (6, 'Charlie', 31, 'DFW', '2023-08-29 00:00:00')
]

schema = StructType([
   StructField("id", IntegerType(), True),
   StructField("name", StringType(), True),
   StructField("age", IntegerType(), True),
   StructField("city", StringType(), True),
   StructField("create_ts", StringType(), True)
])

df = spark.createDataFrame(records, schema)


(
   df.write
   .format("delta")
   .partitionBy("city")
   .save(f"{local_base_path}/{table_name}")
)

[alberttwong]

Now write in delta.

pyspark   --packages com.amazonaws:aws-java-sdk-s3:1.11.271,org.apache.hadoop:hadoop-aws:2.10.2,io.delta:delta-core_2.12:2.1.0   --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension"   --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog" --conf "spark.hadoop.fs.s3a.aws.credentials.provider=org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider"

pyspark script

from pyspark.sql.types import *
from pyspark import SparkContext
from pyspark.sql import SparkSession
import os

# initialize the bucket
database_name = "delta_db"
table_name = "people"
local_base_path = "s3a://warehouse"

records = [
   (1, 'John', 25, 'NYC', '2023-09-28 00:00:00'),
   (2, 'Emily', 30, 'SFO', '2023-09-28 00:00:00'),
   (3, 'Michael', 35, 'ORD', '2023-09-28 00:00:00'),
   (4, 'Andrew', 40, 'NYC', '2023-10-28 00:00:00'),
   (5, 'Bob', 28, 'SEA', '2023-09-23 00:00:00'),
   (6, 'Charlie', 31, 'DFW', '2023-08-29 00:00:00')
]

schema = StructType([
   StructField("id", IntegerType(), True),
   StructField("name", StringType(), True),
   StructField("age", IntegerType(), True),
   StructField("city", StringType(), True),
   StructField("create_ts", StringType(), True)
])

df = spark.createDataFrame(records, schema)


(
   df.write
   .format("delta")
   .partitionBy("city")
   .save(f"{local_base_path}/{table_name}")
)

Traceback (most recent call last):
  File "<stdin>", line 2, in <module>
  File "/spark/python/pyspark/sql/readwriter.py", line 1398, in save
    self._jwrite.save(path)
  File "/spark/python/lib/py4j-0.10.9.7-src.zip/py4j/java_gateway.py", line 1322, in __call__
  File "/spark/python/pyspark/errors/exceptions/captured.py", line 169, in deco
    return f(*a, **kw)
  File "/spark/python/lib/py4j-0.10.9.7-src.zip/py4j/protocol.py", line 326, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o74.save.
: java.lang.NoClassDefFoundError: org/apache/spark/sql/execution/datasources/FileFormatWriter$Empty2Null

[alberttwong]

issue was wrong delta version per https://docs.delta.io/latest/releases.html#compatibility-with-apache-spark

pyspark   --packages com.amazonaws:aws-java-sdk-s3:1.11.271,org.apache.hadoop:hadoop-aws:2.10.2,io.delta:delta-core_2.12:2.4.0   --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension"   --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog" --conf "spark.hadoop.fs.s3a.aws.credentials.provider=org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider"

[alberttwong]

I wrote 6 rows into delta on s3 and then used xtable to covert the data to iceberg and hudi. Next I used run_sync_tool.sh to register the hudi files in HMS and then query it into Trino. No data in Trino.

I don't see any errors.

export AWS_SECRET_ACCESS_KEY=password
export AWS_ACCESS_KEY_ID=admin
export ENDPOINT=http://minio:9000
export AWS_REGION=us-east-1
cd /opt/xtable/jars/; java -jar xtable-utilities-0.1.0-SNAPSHOT-bundled.jar --datasetConfig xtable_delta.yaml -p core-site.xml

[alberttwong]

pyspark --packages org.apache.hudi:hudi-hive-sync-bundle:0.14.1,com.amazonaws:aws-java-sdk-s3:1.11.271,org.apache.hadoop:hadoop-client:2.10.2,org.apache.hadoop:hadoop-aws:2.10.2,org.apache.parquet:parquet-avro:1.10.1,com.esotericsoftware:kryo-shaded:4.0.2

add ~/.ivy2/jars/*: to classpath

vi /opt/hudi/hudi-sync/hudi-hive-sync/run_sync_tool.sh

cd /opt/hudi/hudi-sync/hudi-hive-sync ; \
./run_sync_tool.sh  \
--metastore-uris 'thrift://hive-metastore:9083' \
--partitioned-by city \
--base-path 's3a://warehouse/people' \
--database hudi_db \
--table people \
--sync-mode hms

[alberttwong]

launch Trino

select * from hudi.hudi_db.people;
select * from delta.delta_db.people;
select * from iceberg.iceberg_db.people;

trino@trino:/$ trino
trino> select * from hudi.hudi_db.people;
    -> select * from delta.delta_db.people;
    -> select * from iceberg.iceberg_db.people;
 _hoodie_commit_time | _hoodie_commit_seqno | _hoodie_record_key | _hoodie_part>
---------------------+----------------------+--------------------+------------->
(0 rows)

Query 20240604_233108_00000_ujp5k, FINISHED, 1 node
Splits: 1 total, 1 done (100.00%)
1.36 [0 rows, 0B] [0 rows/s, 0B/s]

 id |  name   | age | city |      create_ts
----+---------+-----+------+---------------------
  5 | Bob     |  28 | SEA  | 2023-09-23 00:00:00
  6 | Charlie |  31 | DFW  | 2023-08-29 00:00:00
  4 | Andrew  |  40 | NYC  | 2023-10-28 00:00:00
  1 | John    |  25 | NYC  | 2023-09-28 00:00:00
  3 | Michael |  35 | ORD  | 2023-09-28 00:00:00
  2 | Emily   |  30 | SFO  | 2023-09-28 00:00:00
(6 rows)

Query 20240604_233113_00001_ujp5k, FINISHED, 1 node
Splits: 6 total, 6 done (100.00%)
0.83 [6 rows, 7.44KB] [7 rows/s, 8.96KB/s]

 id |  name   | age | city |      create_ts
----+---------+-----+------+---------------------
  1 | John    |  25 | NYC  | 2023-09-28 00:00:00
  4 | Andrew  |  40 | NYC  | 2023-10-28 00:00:00
  2 | Emily   |  30 | SFO  | 2023-09-28 00:00:00
  3 | Michael |  35 | ORD  | 2023-09-28 00:00:00
  6 | Charlie |  31 | DFW  | 2023-08-29 00:00:00
  5 | Bob     |  28 | SEA  | 2023-09-23 00:00:00
(6 rows)

Query 20240604_233114_00002_ujp5k, FINISHED, 1 node
Splits: 6 total, 6 done (100.00%)
0.95 [6 rows, 7.44KB] [6 rows/s, 7.8KB/s]

[alberttwong]

no data in Hudi tables. Linked github issue #460.

It looks like there needs to be changes to the Hudi Trino Connector. trinodb/trino#22289

[alberttwong]

advice was to use Trino 418

[alberttwong]

even using Trino 418, I don't get any data back from the query.

trino> select * from hudi.hudi_db.people;
 _hoodie_commit_time | _hoodie_commit_seqno | _hoodie_record_key | _hoodie_partition_path | _hoodie_file_name | id | name | age | create_ts | city 
---------------------+----------------------+--------------------+------------------------+-------------------+----+------+-----+-----------+------
(0 rows)

Query 20240715_192045_00013_5pvwr, FINISHED, 1 node
Splits: 1 total, 1 done (100.00%)
0.18 [0 rows, 0B] [0 rows/s, 0B/s]

trino>