我的目标是从GCS读取一个文件并将其写入Cassandra。对于Apache Beam/Dataflow来说，我是一个新手，我可以找到使用Python构建的大部分方法。不幸的是，CassandraIO对于Beam来说只是Java原生的。
我使用单词计数示例作为模板，并尝试摆脱TextIO.write()，并将其替换为CassandraIO.<Words>write()。
下面是我的Cassandra表的java类

package org.apache.beam.examples;

import java.io.Serializable;
import com.datastax.driver.mapping.annotations.Column;
import com.datastax.driver.mapping.annotations.PartitionKey;
import com.datastax.driver.mapping.annotations.Table;

@Table(keyspace = "test", name = "words", readConsistency = "ONE", writeConsistency = "QUORUM",
        caseSensitiveKeyspace = false, caseSensitiveTable = false)
public class Words implements Serializable {
//    private static final long serialVersionUID = 1L;

    @PartitionKey
    @Column(name = "word")
    public String word;

    @Column(name = "count")
    public long count;

    public Words() {
    }

    public Words(String word, int count) {
        this.word = word;
        this.count = count;
    }

    @Override
    public boolean equals(Object obj) {
        Words other = (Words) obj;
        return this.word.equals(other.word) && this.count == other.count;
    }
}

这里是主代码的管道部分。

static void runWordCount(WordCount.WordCountOptions options) {
        Pipeline p = Pipeline.create(options);

        // Concepts #2 and #3: Our pipeline applies the composite CountWords transform, and passes the
        // static FormatAsTextFn() to the ParDo transform.
        p.apply("ReadLines", TextIO.read().from(options.getInputFile()))
                .apply(new WordCountToCassandra.CountWords())
                
// Here I'm not sure how to transform PCollection<KV> into PCollection<Words> 

.apply(MapElements.into(TypeDescriptor.of(Words.class)).via(PCollection<KV<String, Long>>)
                }))

                .apply(CassandraIO.<Words>write()
                        .withHosts(Collections.singletonList("my_ip"))
                        .withPort(9142)
                        .withKeyspace("test")
                        .withEntity(Words.class));

        p.run().waitUntilFinish();
    }

我的理解是使用一个PTransform从PCollection<T1>传递到PCollection<T2>，我不知道如何Map它。