In order to generate split graphs from the ttl dump, we need to orchestrate spark jobs for data preparation and transfer.
Ultimately we need to generate snapshots in n3 format that will be ingested by the T428235: [NEEDS GROOMING] data-reload: create a QLever index using an ephemeral instance.

The following WDQS v1 steps needs to be implemented in the WDQS v2 data pipeline (airflow dag).

• T429254: data-reload: keep data pipeline tables DDL under version control
• T429248: data-reload: munge_dumps task reads TTL data from S3 and writes to HDFS
• T429250: data-reload: implement gen_rev_map task
• T429252: data-relaod: implement split_subgraphs task
• T429253: data-reload: implement export_n3 task

munge_dumps can read the ttl file we generate (on k8s) from S3, and store its output in HDFS for it's children task to pickup.
The output of export_n3 should be store both on HDFS / Hive for analytics, as well as S3 for exposing it to the qlever-indexer file.

Modulo changes in input / output paths and tables that need to be stored in the wikidata database, not business logic changes are expected. We need to benchmark i/o perf when reading/writing for s3.

Spark execution

Ideally we want to execute the spark workload on k8s, but a first iteration would work fine on hadoop (hdfs and hive metastore). We want to decouple the business logic (our spark jobs and hive tables) from the execution infra (hadoop or k8s and the respective airflow operators).

This approach might require some sync with DE (cc @JAllemandou @Ahoelzl ) since AFAIK we don't have any workflow writing to s3 from hadoop yet.

Spark (Hadoop) has been tested and hadoop workers can access s3. We can start implementing with the current hadoop target, and swap for spark-on-k8s when ready.

@JAllemandou I tested interactively with a dummy job:

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("s3-connectivity-test") \
    .getOrCreate()

data = [("hello", 1), ("world", 2)]
df = spark.createDataFrame(data, ["word", "count"])

df.write \
    .mode("overwrite") \
    .parquet("s3a://gmodena-test-bucket.dse-k8s-eqiad/spark-out")

print("Write succeeded!")
spark.stop()

Launched with

spark3-submit -v \
  --jars hadoop-aws-3.2.0.jar,aws-java-sdk-bundle-1.11.888.jar \
  --conf "spark.hadoop.fs.s3a.endpoint=rgw.eqiad.dpe.anycast.wmnet" \
  --conf "spark.hadoop.fs.s3a.access.key=${s3_access_key}" \
  --conf "spark.hadoop.fs.s3a.secret.key=${s3_secret_key}" \
  --conf "spark.hadoop.fs.s3a.path.style.access=true" \
  --conf "spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem" \
  s3_test.py

This bring in dependencies on hadoop-aws-3.2.0.jar,aws-java-sdk-bundle-1.11.888.jar, but they are Apache 2 licensed and available in maven central.

We have s3 credentials already available in k8s and airflow (via a connection setting), but we might need to patch the ApacheSparkSubmitOperator to pick them up (or provide custom properties). If you are comfortable with this direction we can pick up the work.