Hi Shiao-An, With regard to your set-up below and I quote:
"The input/output files are parquet on GCS. The Spark version is 2.4.4 with standalone deployment. Workers running on GCP preemptible instances and they being preempted very frequently." Am I correct that you have foregone deploying Dataproc clusters on GCP in favour of selecting some VM boxes, installing your own Spark cluster running Spark in standalone mode (assuming to save costs $$$). What is the rationale behind this choice? Accordingly <https://cloud.google.com/compute/docs/instances/preemptible> *Compute Engine might stop (preempt) these instances if it requires access to those resources for other tasks. Preemptible instances are excess Compute Engine capacity, so their availability varies with usage*. So what causes some VM instances to be preempted? I have not touched standalone mode for couple of years myself. So your ETL process reads the raw snapshots, does some joins and creates new hourly processed snapshots. There seems to be not an intermediate stage to verify the sanity of data (data lineage). Personally I would deploy a database to do this ETL. That would give you an option to look at your data easier and store everything in a staging area before final push to the analytics layer. HTH LinkedIn * https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw <https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw>* *Disclaimer:* Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction. On Tue, 29 Dec 2020 at 14:01, Shiao-An Yuan <shiao.an.y...@gmail.com> wrote: > Hi folks, > > We recently identified a data correctness issue in our pipeline. > > The data processing flow is as follows: > 1. read the current snapshot (provide empty if it doesn't exist yet) > 2. read unprocessed new data > 3. union them and do a `reduceByKey` operation > 4. output a new version of the snapshot > 5. repeat step 1~4 > > The simplified version of code: > ``` > // schema > case class Log(pkey: Array[Byte], a: String, b: Int, /* 100+ columns */) > > // function for reduce > def merge(left: Log, right: Log): Log = { > Log(pkey = left.pkey > a = if (left.a!=null) left.a else right.a, > b = if (left.a!=null) left.b else right.b, > ... > ) > } > > // a very large parquet file (>10G, 200 partitions) > val currentSnapshot = spark.read.schema(schema).parquet(...).as[Log] > > // multiple small parquet files > val newAddedLogs = spark.read.schema(schema).parquet(...).as[Log] > > val newSnapshot = currentSnapshot.union(newAddedLog) > .groupByKey(new String(pkey)) // generate key > .reduceGroups(_.merge(_)) // > spark.sql.shuffle.partitions=200 > .map(_._2) // drop key > > newSnapshot > .repartition(60) // (1) > .write.parquet(newPath) > ``` > > The issue we have is that some data were duplicated or lost, and the > amount of > duplicated and loss data are similar. > > We also noticed that this situation only happens if some instances got > preempted. Spark will retry the stage, so some of the partitioned files are > generated at the 1st time, and other files are generated at the 2nd(retry) > time. > Moreover, those duplicated logs will be duplicated exactly twice and > located in > both batches (one in the first batch; and one in the second batch). > > The input/output files are parquet on GCS. The Spark version is 2.4.4 with > standalone deployment. Workers running on GCP preemptible instances and > they > being preempted very frequently. > > The pipeline is running in a single long-running process with > multi-threads, > each snapshot represent an "hour" of data, and we do the > "read-reduce-write" operations > on multiple snapshots(hours) simultaneously. We pretty sure the same > snapshot(hour) never process parallelly and the output path always > generated with a timestamp, so those jobs shouldn't affect each other. > > After changing the line (1) to `coalesce` or `repartition(100, $"pkey")` > the issue > was gone, but I believe there is still a correctness bug that hasn't been > reported yet. > > We have tried to reproduce this bug on a smaller scale but haven't > succeeded yet. I > have read SPARK-23207 and SPARK-28699, but couldn't found the bug. > Since this case is DataSet, I believe it is unrelated to SPARK-24243. > > Can anyone give me some advice about the following tasks? > Thanks in advance. > > Shiao-An Yuan >
