如何在 2.2.0 中获取给定 Apache Spark Dataframe 的 Cassandra cql 字符串？

Question

我正在尝试获取给定 Dataframe 的 cql 字符串。我遇到了这个function

我可以在哪里做这样的事情

TableDef.fromDataFrame(df, "test", "hello", ProtocolVersion.NEWEST_SUPPORTED).cql()

在我看来，该库使用第一列作为分区键并且不关心集群键，那么我如何指定使用 Dataframe 的特定列集作为 PartitionKey 和 ParticularSet 列作为集群键？

看起来我可以创建一个新的 TableDef，但是我必须自己完成整个映射，并且在某些情况下，Java 中无法访问像 ColumnType 这样的必要函数。例如，我尝试创建一个新的 ColumnDef，如下所示

new ColumnDef("col5", new PartitionKeyColumn(), ColumnType is not accessible in Java)

目标：从 Spark DataFrame 中获取 CQL 创建语句。

输入 我的数据框可以有任意数量的列以及它们各自的 Spark 类型。所以说我有一个包含 100 列的 Spark 数据框，其中我的数据框的 col8、col9 对应于 cassandra partitionKey 列，我的 column10 对应于 cassandra 聚类键列

col1| col2| ...|col100

现在我想使用 spark-cassandra-connector 库给我一个 CQL 创建表语句给定上面的信息。

期望的输出

create table if not exists test.hello (
   col1 bigint, (whatever column1 type is from my dataframe I just picked bigint randomly)
   col2 varchar,
   col3 double,
   ...
   ...
   col100 bigint,
   primary key(col8,col9)
) WITH CLUSTERING ORDER BY (col10 DESC);

Answer 1

由于必需的组件（PartitionKeyColumn 和ColumnType 的实例）是 Scala 中的对象，因此您需要使用以下语法来访问它们的实例：

// imports
import com.datastax.spark.connector.cql.ColumnDef;
import com.datastax.spark.connector.cql.PartitionKeyColumn$;
import com.datastax.spark.connector.types.TextType$;

// actual code
ColumnDef a = new ColumnDef("col5",  
      PartitionKeyColumn$.MODULE$, TextType$.MODULE$);

查看ColumnRole 和PrimitiveTypes 的代码以查找对象/类名称的完整列表。

附加要求后更新：代码很长，但应该可以工作...

SparkSession spark = SparkSession.builder()
                .appName("Java Spark SQL example").getOrCreate();

Set<String> partitionKeys = new TreeSet<String>() {{
                add("col1");
                add("col2");
        }};
Map<String, Integer> clustereingKeys = new TreeMap<String, Integer>() {{
                put("col8", 0);
                put("col9", 1);
        }};

Dataset<Row> df = spark.read().json("my-test-file.json");
TableDef td = TableDef.fromDataFrame(df, "test", "hello", 
                ProtocolVersion.NEWEST_SUPPORTED);

List<ColumnDef> partKeyList = new ArrayList<ColumnDef>();
List<ColumnDef> clusterColumnList = new ArrayList<ColumnDef>();
List<ColumnDef> regColulmnList = new ArrayList<ColumnDef>();

scala.collection.Iterator<ColumnDef> iter = td.allColumns().iterator();
while (iter.hasNext()) {
        ColumnDef col = iter.next();
        String colName = col.columnName();
        if (partitionKeys.contains(colName)) {
                partKeyList.add(new ColumnDef(colName, 
                                PartitionKeyColumn$.MODULE$, col.columnType()));
        } else if (clustereingKeys.containsKey(colName)) {
                int idx = clustereingKeys.get(colName);
                clusterColumnList.add(new ColumnDef(colName, 
                                new ClusteringColumn(idx), col.columnType()));
        } else {
                regColulmnList.add(new ColumnDef(colName, 
                                RegularColumn$.MODULE$, col.columnType()));
        }
}

TableDef newTd = new TableDef(td.keyspaceName(), td.tableName(), 
                (scala.collection.Seq<ColumnDef>) partKeyList,
                (scala.collection.Seq<ColumnDef>) clusterColumnList, 
                (scala.collection.Seq<ColumnDef>) regColulmnList,
                td.indexes(), td.isView());
String cql = newTd.cql();
System.out.println(cql);