Merge algorithm with upstream.

Makefile

+venv: requirements.txt
+	test -d venv || virtualenv --python=$(shell which python3) venv
+	. venv/bin/activate; pip install -Ur requirements.txt;
+
+test:	venv
+	. venv/bin/activate; pytest --cov etl

conftest.py

+import pytest
+from etl.transform import RawDataset
+
+
+@pytest.fixture(scope="session")
+def raw_data(spark_session):
+    return spark_session.createDataFrame(
+        [
+            (
+                "0",
+                "Q1234",
+                "44444",
+                "Some page with suggestions",
+                '[{"image": "image1.jpg", "rating": 2.0, "note": "image was found in the following Wikis: ruwiki"}]',
+                "arwiki",
+                "2020-12",
+            ),
+            (
+                "1",
+                "Q56789",
+                "55555",
+                "Some page with no suggestion",
+                None,
+                "arwiki",
+                "2020-12"
+            )
+        ],
+        RawDataset.schema,
+    )

ddl/external_imagerec_prod.hql

@@ -24,7 +24,8 @@ ROW FORMAT SERDE
   'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
 WITH SERDEPROPERTIES (
   'field.delim'='\t',
-  'serialization.format'='\t')
+  'serialization.format'='\t',
+  'serialization.null.format'='""')
 STORED AS INPUTFORMAT
   'org.apache.hadoop.mapred.TextInputFormat'
 OUTPUTFORMAT
@@ -32,5 +33,6 @@ OUTPUTFORMAT
 LOCATION
   'hdfs://analytics-hadoop/user/${hiveconf:username}/imagerec_prod/data';
 
+
 -- Update partition metadata
 MSCK REPAIR TABLE `imagerec_prod`;

etl/transform.py

@@ -52,8 +52,9 @@ class ImageRecommendation:
             )
 
     def transform(self) -> DataFrame:
-        return (
-            self.dataFrame.withColumn(
+        with_recommendations = (
+            self.dataFrame.where(~F.col("top_candidates").isNull())
+            .withColumn(
                 "data",
                 F.explode(
                     F.from_json("top_candidates", RawDataset.recommendation_schema)
@@ -73,6 +74,22 @@ class ImageRecommendation:
                 "source",
             )
         )
+        without_recommendations = (
+            self.dataFrame.where(F.col("top_candidates").isNull())
+            .withColumnRenamed("wiki_db", "wiki")
+            .withColumn("image_id", F.lit(None))
+            .withColumn("confidence_rating", F.lit(None))
+            .withColumn("source", F.lit(None))
+            .select(
+                "wiki",
+                "page_id",
+                "page_title",
+                "image_id",
+                "confidence_rating",
+                "source",
+            )
+        )
+        return with_recommendations.union(without_recommendations)
 
 
 if __name__ == "__main__":

requirements.txt

+pytest==6.2.2
+pytest-spark==0.6.0
+pytest-cov==2.10.1
\ No newline at end of file

tests/test_etl.py

+from etl.transform import RawDataset, ImageRecommendation
+
+
+def test_etl(raw_data):
+    assert raw_data.count() == 2
+
+    ddf = ImageRecommendation(raw_data).transform()
+    assert (
+        len(
+            set(ddf.columns).difference(
+                {
+                    "wiki",
+                    "page_id",
+                    "page_title",
+                    "image_id",
+                    "confidence_rating",
+                    "source",
+                }
+            )
+        )
+        == 0
+    )
+
+    expected_num_records = 2
+    assert ddf.count() == expected_num_records