用Spark验证CSV文件列

Question

我试图在Spark中读取一个CSV文件(应该有一个头),并将数据加载到一个现有的表中(带有预定义的列和数据类型)。csv文件可能非常大，所以如果csv的列标题不是“有效”的，我可以避免这样做，那就太好了。

当我当前读取文件时，我指定StructType作为模式，但这并不能验证标头是否包含正确顺序的正确列。这是我到目前为止所做的（我正在另一个地方构建“schema”结构类型）：

sqlContext
  .read()
  .format("csv")
  .schema(schema)
  .load("pathToFile");

如果我添加<代码>。option("header "，" true)" line它将跳过csv文件的第一行，并使用我在StructType的< code>add方法中传递的名称。(例如，如果我用“id”和“name”构建StructType，并且csv中的第一行是“idzzz，name”，那么得到的dataframe将有“id”和“name”列。我希望能够验证csv标题与我计划加载csv的表具有相同的列名。

我尝试用< code >读取文件。head()，并对第一行进行一些检查，但这样会下载整个文件。

任何建议都非常欢迎。

Answer 1

据我所知，您希望验证您读取的CSV的模式。schema选项的问题在于，它的目标是告诉spark它是数据的模式，而不是检查它是否是。

但是，有一个选项可以在读取CSV时推断所述架构，并且在您的情况下可能非常有用（推断Schema）。然后，您可以将该架构与您期望的等式架构进行比较，或者执行我将介绍的小解决方法，使其更加宽松。

让我们看看它是如何工作的以下文件：

a,b
1,abcd
2,efgh

然后，让我们读数据。我使用了scala REPL，但您应该能够非常轻松地将所有这些转换为Java。

val df = spark.read
    .option("header", true) // reading the header
    .option("inferSchema", true) // infering the sschema
    .csv(".../file.csv")
// then let's define the schema you would expect
val schema = StructType(Array(StructField("a", IntegerType),
                              StructField("b", StringType)))

// And we can check that the schema spark inferred is the same as the one
// we expect:
schema.equals(df.schema)
// res14: Boolean = true

更进一步

这是一个完美的世界。事实上，如果您的模式包含例如不可为空的列或其他小差异，这种基于对象严格相等的解决方案将不起作用。

val schema2 = StructType(Array(StructField("a", IntegerType, false),
                               StructField("b", StringType, true)))
// the first column is non nullable, it does not work because all the columns
// are  nullable when inferred by spark:
schema2.equals(df.schema)
// res15: Boolean = false

在这种情况下，您可能需要实现适合您的模式比较方法：

def equalSchemas(s1 : StructType, s2 : StructType) = {
  s1.indices
    .map(i => s1(i).name.toUpperCase.equals(s2(i).name.toUpperCase) &&
              s1(i).dataType.equals(s2(i).dataType))
    .reduce(_ && _)
}
equalSchemas(schema2, df.schema)
// res23: Boolean = true

我正在检查列的名称和类型是否匹配，以及顺序是否相同。您可能需要根据需要实现不同的逻辑。