revans2
commented
1 year ago It should be noted that something similar to this already exists in the tools repo.
https://github.com/NVIDIA/spark-rapids-tools/tree/dev/data_validation
I'm not 100% sure if we want to have two tools or just one. But the code is not that complicated. The data_validation tool is not great. It requires a primary key, which should not be a requirement. It does not handle maps, nulls, duplicate rows, or floating point approximate matches correctly.
The above example can be updated to handle nulls by using the null equals operator '<=>' as the join condition for the anti-join columns. Maps can be replaced with a map_entries that are array_sorted. And for now we are just going to punt on floats/doubles in both cases and not include them in our scale testing suite until this can support doing a fuzzy check for them.
wjxiz1992
Is your feature request related to a problem? Please describe. https://github.com/NVIDIA/spark-rapids/pull/8706 adds a tool to generate large data sets for scale testing, and possibly some performance testing for very difficult corner cases. But ultimately we need to also verify that the results after running that generated data through a query are correct. TPC- benchmarks provide a validation tool that can be used to check if the results are correct, but for us, our definition of correct is that it matches Spark on the CPU. Warts and all.
The other problem is that the data produced by scale testing can be very very large (both in terms of rows, but also column count, and nesting). So we need to come up with a way to validate that the results are correct in a way that is going to scale. This itself may be a really massive scale test too because we are likely to double the number of rows and columns by trying to compare two large data sets.
One way to do this is to group by all of the columns, do a count on them (so that all duplicates are removed), for both the expected data set and the test data set. Then to do two anti-joins on them. A left-anti-join to know if there are rows on the LHS that are not in the right, and a right-anti-join to know if there are rows in the RHS that are not in the left.
This works if all of the values produced will be bit for bit identical to those on the CPU, and are types that Spark can join on. For floating point types, especially with aggregations that is not guaranteed. Perhaps that is good enough. We just avoid floats in our scale tests and we are good, but I am not sure we can do that especially if we want to reproduce issues from customer queries.
We might need to normalize maps to make this work and turn them into a combinations of structs and arrays with the data sorted by the key.
For floating point values or anywhere that we would want to do an approximate match, I just don't know of a good way to do it that is distributed and scales well.
If we try to normalize the floating point values by rounding of any kind there are inflection points where if we are off by a little the rounding would just exacerbate the difference. We could provide a UDF to do a fuzzy match on the floating point values as a part of the anti-join, but we could not do the group by any more to deal with duplicates because we might not even be self consistent, and aggregations don't support a fuzzy match.
The only option I can think of is to sort all the data from both data sets and produce a row number. This takes a bit of work with monotonically_increasing_id and partition_id, along with writing out the intermediate data, but it is possible. After that we can join on the row numbers and do a fuzzy match in a UDF for each column that contains floating point values, or maps. Not ideal but it would work.