scala - 我需要将配置单元表架构与包含csv文件架构的数据框进行比较

Question

在这里，我试图验证配置单元表和s3中存储的CSV文件的结构。

这是数据框中的CSV文件的架构。

+----+----------------+----+---+-----------+-----------+
|S_No|        Variable|Type|Len|     Format|   Informat|
+----+----------------+----+---+-----------+-----------+
|   1|        DATETIME| Num|  8|DATETIME20.|DATETIME20.|
|   2|   LOAD_DATETIME| Num|  8|DATETIME20.|DATETIME20.|
|   3|     SOURCE_BANK|Char| 1 |       null|       null|
|   4|        EMP_NAME|Char| 50|       null|       null|
|   5|HEADER_ROW_COUNT| Num|  8|       null|       null|
|   6|      EMP _HOURS| Num|  8|       15.2|       15.1|
+----+----------------+----+---+-----------+-----------+

我需要将其与O / p进行比较

import org.apache.spark.sql.hive.HiveContext
val targetTableName = "TableA"
val hc = new HiveContext(sc)
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
val targetRawData = hc.sql("Select datetime,load_datetime,trim(source_bank) as source_bank,trim(emp_name) as emp_name,header_row_count, emp_hours from " + targetTableName)

val schema= targetRawData.schema

:schema: org.apache.spark.sql.types.StructType = StructType(StructField(datetime,TimestampType,true), StructField(load_datetime,TimestampType,true), StructField(source_bank,StringType,true), StructField(emp_name,StringType,true), StructField(header_row_count,IntegerType,true), StructField(emp_hours,DoubleType,true))

Answer 1

您也可以使用MegaSparDiff开放源代码，该源代码可以比较多种类型的数据源，包括S3，HPE，CSV，JDBC等。

https://github.com/FINRAOS/MegaSparkDiff

下面的对将作为数据帧返回inLeftButNotInRight和inRightButNotInLeft。

您的成功条件是，如果两个DataFrame的记录都为零，则意味着数据是相同的。
现在您可能想要考虑不加载所有数据，因为您只对比较模式感兴趣。

 SparkFactory.initializeSparkContext();

    AppleTable leftAppleTable = SparkFactory.parallelizeTextSource("S3://file1","table1");

    AppleTable rightAppleTable = SparkFactory.parallelizeHiveSource("select * from hivetable","hivetable");

    Pair<Dataset<Row>, Dataset<Row>> resultPair = SparkCompare.compareAppleTables(leftAppleTable, rightAppleTable);

    if (resultPair.getLeft().count() != 0 && resultPair.getRight().count() != 0)
    {
        //success condition
    }

    SparkFactory.stopSparkContext();