nsivabalan commented on code in PR #20496: URL: https://github.com/apache/hudi/pull/20496#discussion_r4452610798
########## hudi-integ-test/spark-pipelines/keyed_flows.py: ########## @@ -0,0 +1,133 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Check keyed Hudi SDP flows across ordering, deletion, reinsertion and restart.""" +import argparse +import json +from pathlib import Path +import sys +import tempfile + +SOURCE = 'from pyspark import pipelines as dp\nfrom pyspark.sql import SparkSession\nspark = SparkSession.active()\[email protected]_view(name="input_events")\ndef input_events():\n return spark.readStream.schema("id INT, ts LONG, part STRING, value STRING, _hoodie_is_deleted BOOLEAN").json(spark.conf.get("hudi.test.input"))\n' +PROPERTIES = "'type'='cow', 'primaryKey'='id', 'preCombineField'='ts', 'hoodie.record.merge.mode'='EVENT_TIME_ORDERING'" +TABLES = ("keyed_new", "keyed_existing") + + +def client(mode, remote, root): + from pyspark.sql import SparkSession + spark = SparkSession.builder.remote(remote).getOrCreate() + try: + assert spark.conf.get("spark.hudi.test.run.id") == root.name + if mode == "prepare": + spark.sql(f"CREATE TABLE sdp_test.keyed_existing (id INT, ts BIGINT, part STRING, value STRING, " + f"_hoodie_is_deleted BOOLEAN) USING hudi PARTITIONED BY (part) TBLPROPERTIES ({PROPERTIES})").collect() + else: + expected = json.loads((root / "expected.json").read_text()) + for table in TABLES: + actual = [list(row) for row in spark.table(f"sdp_test.{table}") + .select("id", "ts", "part", "value").orderBy("id").collect()] + assert actual == expected, (table, actual, expected) + print(f"PASS {mode} {table}: {actual}") + finally: + spark.stop() + + +def scenario(h): + source = h.root / "input" + source.mkdir() + (h.root / "source.py").write_text(SOURCE) + flows = "SET spark.hoodie.datasource.write.operation=upsert;\n" + "\n".join( + f"CREATE STREAMING TABLE {table} USING hudi PARTITIONED BY (part) TBLPROPERTIES ({PROPERTIES}) " + "AS SELECT id, ts, part, value, _hoodie_is_deleted FROM STREAM input_events;" for table in TABLES) + (h.root / "flows.sql").write_text(flows) + spec = {"name": "keyed_flows", "storage": (h.root / "checkpoints").as_uri(), + "catalog": "spark_catalog", "database": "sdp_test", + "libraries": [{"glob": {"include": "source.py"}}, {"glob": {"include": "flows.sql"}}], + "configuration": {"hudi.test.input": source.as_uri(), "spark.sql.shuffle.partitions": "2", + "spark.hoodie.metadata.enable": "false", + "spark.sql.pipelines.maxFlowRetryAttempts": "0"}} + path = h.root / "spark-pipeline.yml" + path.write_text(json.dumps(spec, indent=2)) + for name in ("source.py", "flows.sql", path.name): + (h.logs / name).write_text((h.root / name).read_text()) + h.client(Path(__file__).resolve(), "prepare") + stages = ( + ([(1, 1, "a", "old", False), (1, 2, "a", "new", False), (2, 1, "b", "keep", False)], + [[1, 2, "a", "new"], [2, 1, "b", "keep"]]), + ([(1, 3, "a", "updated", False)], [[1, 3, "a", "updated"], [2, 1, "b", "keep"]]), + ([(1, 2, "a", "late", False)], [[1, 3, "a", "updated"], [2, 1, "b", "keep"]]), + ([(1, 4, "a", "deleted", True)], [[2, 1, "b", "keep"]]), + ([(1, 5, "a", "reinserted", False)], [[1, 5, "a", "reinserted"], [2, 1, "b", "keep"]])) + for i, (rows, expected) in enumerate(stages, 1): + (source / f"{i}.json").write_text("".join(json.dumps(dict(zip( + ("id", "ts", "part", "value", "_hoodie_is_deleted"), row))) + "\n" for row in rows)) + (h.root / "expected.json").write_text(json.dumps(expected)) + h.run_pipeline(path, f"stage-{i}") + h.client(Path(__file__).resolve(), f"verify-{i}") + h.run_pipeline(path, "restart") + h.client(Path(__file__).resolve(), "verify-restart") + invalid = ( + ("missing_key", PROPERTIES.replace("'primaryKey'='id'", "'primaryKey'='missing_key'"), "part", SOURCE, "missing_key"), + ("missing_order", PROPERTIES.replace("'preCombineField'='ts'", "'preCombineField'='missing_order'"), "part", SOURCE, "missing_order"), + ("missing_partition", PROPERTIES, "missing_partition", SOURCE, "missing_partition"), + ("invalid_marker", PROPERTIES, "part", SOURCE.replace("_hoodie_is_deleted BOOLEAN", "_hoodie_is_deleted STRING"), "_hoodie_is_deleted"), + ("changed_identity", PROPERTIES.replace("'primaryKey'='id'", "'primaryKey'='value'"), "part", SOURCE, "recordkey")) Review Comment: `error_text = "recordkey"` for `changed_identity` matches almost any Hudi log. Could this assert the actual rejection message for a changed record key instead? ########## hudi-integ-test/spark-pipelines/keyed_flows.py: ########## @@ -0,0 +1,133 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Check keyed Hudi SDP flows across ordering, deletion, reinsertion and restart.""" +import argparse +import json +from pathlib import Path +import sys +import tempfile + +SOURCE = 'from pyspark import pipelines as dp\nfrom pyspark.sql import SparkSession\nspark = SparkSession.active()\[email protected]_view(name="input_events")\ndef input_events():\n return spark.readStream.schema("id INT, ts LONG, part STRING, value STRING, _hoodie_is_deleted BOOLEAN").json(spark.conf.get("hudi.test.input"))\n' +PROPERTIES = "'type'='cow', 'primaryKey'='id', 'preCombineField'='ts', 'hoodie.record.merge.mode'='EVENT_TIME_ORDERING'" +TABLES = ("keyed_new", "keyed_existing") + + +def client(mode, remote, root): + from pyspark.sql import SparkSession + spark = SparkSession.builder.remote(remote).getOrCreate() + try: + assert spark.conf.get("spark.hudi.test.run.id") == root.name + if mode == "prepare": + spark.sql(f"CREATE TABLE sdp_test.keyed_existing (id INT, ts BIGINT, part STRING, value STRING, " + f"_hoodie_is_deleted BOOLEAN) USING hudi PARTITIONED BY (part) TBLPROPERTIES ({PROPERTIES})").collect() + else: + expected = json.loads((root / "expected.json").read_text()) + for table in TABLES: + actual = [list(row) for row in spark.table(f"sdp_test.{table}") + .select("id", "ts", "part", "value").orderBy("id").collect()] + assert actual == expected, (table, actual, expected) + print(f"PASS {mode} {table}: {actual}") + finally: + spark.stop() + + +def scenario(h): + source = h.root / "input" + source.mkdir() + (h.root / "source.py").write_text(SOURCE) + flows = "SET spark.hoodie.datasource.write.operation=upsert;\n" + "\n".join( Review Comment: Why is `SET spark.hoodie.datasource.write.operation=upsert` needed? On master the streaming sink already defaults to `UPSERT`. If the scenario fails without this `SET`, keyed SDP streaming tables do not actually default to upsert, and users would silently get duplicates. That would need a fix or at least documentation. Could you drop the `SET`, or explain why it is needed? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
