Personalized dev environment on analytics cluster with Airflow setup (stat1006) - was able to execute job, slightly hacked up to get specific dates and not keep running regularly (eats lots of disk) to get dr0ptp4kt.wikibase_rdf_with_split using my Kerberos principal. Verifying Jupyter notebook approach from David / Andy on stat1005 - some glitches as to be expected, but worked okay by doubling timeouts and removing some caps. Next up, working on a job that will do the splitting in a fashion similar to what's achieved with the join-antijoin approach of the notebooks. I'll want to have the produced data separated out from the existing table, I think - in this case it would be okay in my opinion to use some extra disk.

TL;DR this is about 45% done.

This week I was working to address non-performant, often hanging or crashy, Spark runs. Last night I managed to get this running better, producing a reduction (the equivalent of val_triples_only_used_by_sas from https://people.wikimedia.org/~andrewtavis-wmde/T342111_spark_sa_subgraph_metrics.html ) in 8 minutes in one pass - instead of 3 hours or, worse, something longer followed by an indefinite hang or crash.

The key here was a couple things. First, higher resource limits (this seems obvious, but isn't always true) and attempting to prevent Spark from broadcast joins (it still tries to do them based on the Spark web UI's DAGs, but doesn't seem to do them at bad times, at least).

"spark.driver.memory": "16g",
"spark.driver.cores": 2,
"spark.executor.memory": "12g",
"spark.executor.cores": 4,
"spark.executor.memoryOverhead": "4g",
"spark.sql.shuffle.partitions": 512,
'spark.dynamicAllocation.maxExecutors': 128,
'spark.locality.wait': '1s', # test 0
'spark.sql.autoBroadcastJoinThreshold': -1

Second, removal of cache() calls and setting some join tables as their own DataFrames. This means likely in practice more disk-based merge behavior on the executors for huge joins, but it works better. I'm interested to explore bucketing as an optimization strategy, but may forego this for production of the table as it doesn't seem necessary at the moment - it may however be useful for the produced table for people doing further join operations so am thinking about this.

I had the small reduction pushing to a Parquet directory in HDFS last night. I will be working to see how performant and reliable pushing a larger data set is and will report back here. From there I'll port from Python to Scala.

It took about 26min 24s to write S_direct_triples (7_293_925_470 rows) in basic Parquet. It's not all the rows (not even for its own partition, as that will include Value and Reference triples as well), but this means it ought to be possible for the job to write total 15B rows with about an hour of wall time (maybe double that to play it safe).

dr0ptp4kt opened https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/529

Draft: Split the graph - DAG

Update: it seems to be working. Thus, I'd say this is maybe 75% complete.

It takes about 1h40m to run and generate the two different partitions.

WIP/Draft patches posted at https://gerrit.wikimedia.org/r/c/wikidata/query/rdf/+/969229 and ^ . They require some refactoring and introduction of tests, and probably some extra config variables - I'll connect with Joseph about that last part.

David, Erik, and I spoke through things earlier today while I opened the repos in my IDE. I'll request code review so I can iterate on this.

Here's what I saw after re-running. So, we should be good with the latest patchset that goes without distinct() on the final graphs.

Without distinct() on final graphs - 1h48m
[dr0ptp4kt.wikibase_rdf_scholarly_split_refactor_no_distinct_less_cache]
scholarly_articles: 7_643_858_365, wikidata_main: 7_677_112_695

With distinct() on final graphs - 1h55m
[dr0ptp4kt.wikibase_rdf_scholarly_split_refactor_using_distinct_less_cache]
scholarly_articles: 7_643_858_365, wikidata_main: 7_677_112_695

During discussion today we figured we should be able to merge, then do a JAR release, then update the Airflow patch to use the right version, and get this onto the Search Airflow instance. So, will tentatively plan for that for tomorrow.

Spark patch merged, new Jenkins build of the rdf JAR done, Airflow patch merged. This is deployed to Search's Airflow instance and the job is running. Thank you, @dcausse and @EBernhardson.

Here's the location of stuff for this job that's currently running.

--deploy-mode cluster hdfs:///wmf/cache/artifacts/airflow/search/rdf-spark-tools-0.3.137-jar-with-dependencies.jar
--input-table-partition-spec discovery.wikibase_rdf_t337013/date=20231016/wiki=wikidata
--output-table-partition-spec discovery.wikibase_rdf_scholarly_split/snapshot=20231016/wiki=wikidata
max_attempts: 1

The job completed. The counts match up on this productionized job compared with the prior one run in my namespace. Following are some Hive queries in case needed later. Below that is a really small sample of the resultant data in tabular format for each partition.

Counts

select count(1) from discovery.wikibase_rdf_scholarly_split where snapshot = '20231016' and wiki = 'wikidata' and
scope = 'scholarly_articles';
7643858365

select count(1) from discovery.wikibase_rdf_scholarly_split where snapshot = '20231016' and wiki = 'wikidata' and
scope = 'wikidata_main';
7677112695

Samples

Note: because the target sample size is so small, it's actually possible to get slightly less than the target number of records due to sparseness in a randomly selected set. One can compensate by setting the numerator higher or the denominator lower if one likes to reduce the possibility of such potential artifacts (e.g., to avoid getting 27 records when one really wants 30 records; below we get 30 records apiece, mind you). Note the horizontal scrollbars at the bottom of the tabular data in case the tables overflow on one's browser's settings in Phabricator (mine do).

EDIT: note that there are < and > symbols in the cell values for URIs. (see the Phab source for reference - I won't run the query again, but it would be possible to concatenate to code format or encode these symbols so they render in a table context in Phab).

select "| " || concat_ws(" | ", subject, predicate, object, context) from discovery.wikibase_rdf_scholarly_split where snapshot = '20231016' and wiki = 'wikidata' and
scope = 'scholarly_articles' and rand() <= (30/7643858365) distribute by rand() sort by rand() limit 30;

select "| " || concat_ws(" | ", subject, predicate, object, context) from discovery.wikibase_rdf_scholarly_split where snapshot = '20231016' and wiki = 'wikidata' and
scope = 'wikidata_main' and rand() <= (30/7677112695) distribute by rand() sort by rand() limit 30;