如何使用 Spark DataFrames 和 Cassandra 设置命名策略答案

【问题标题】：How to set naming strategy using Spark DataFrames and Cassandra如何使用 Spark DataFrames 和 Cassandra 设置命名策略
【发布时间】：2018-08-23 23:48:27
【问题描述】：

我有使用 lowerCamelCase 变量名称约定的案例类。 Cassandra 对行和表名使用 lower_underscore 约定。有没有办法指示 DataFrame 自动转换名称？

表定义：

CREATE TABLE data_storage.dummy (
  some_name text,
  some_value text,
  PRIMARY KEY (some_name)
);

火花代码：

val ds = List(Dummy("key", 1)).toDS()

ds.write
    .format("org.apache.spark.sql.cassandra")
    .option("keyspace", "data_storage")
    .option("table", "dummy")
    .save

我编写了一个函数，将名称从驼峰更改为下划线，反之亦然，但在一些遗留案例类也使用 lower_underscore 命名约定的情况下，它不起作用。

  private def toUnderscore(ds: DataFrame): DataFrame = {
    normalizeNames(ds, CaseFormat.LOWER_CAMEL, CaseFormat.LOWER_UNDERSCORE)
  }

  private def toCamel(ds: DataFrame): DataFrame =
    normalizeNames(ds, CaseFormat.LOWER_UNDERSCORE, CaseFormat.LOWER_CAMEL)

  private def normalizeNames(df: DataFrame, from: CaseFormat, to: CaseFormat): DataFrame = {
    def normalizeName(c: String): String = from.to(to, c.toString)

    def flattenSchema(schema: StructType): StructType = {
      StructType(schema.fields.map {
        case StructField(name, inner: StructType, nullable, metadata) =>
          StructField(normalizeName(name), flattenSchema(inner), nullable, metadata)
        case StructField(name, array: ArrayType, nullable, metadata)  =>
          StructField(
            normalizeName(name), array.copy(elementType = array.elementType match {
              case s: StructType => flattenSchema(s)
              case x             => x
            }), nullable, metadata)
        case StructField(name, flat, nullable, metadata)              =>
          StructField(normalizeName(name), flat, nullable, metadata)
      })
    }

【问题讨论】：

标签： scala apache-spark cassandra spark-dataframe

【解决方案1】：

这种名称转换没有内置方法。 RDD 接口围绕它的类型转换有一些额外的方法来执行这些类型的转换，但通常我们没有为 DataSets/DataFrames 添加机制。

【讨论】：