从Spark导入长行到Cassandra

Question

我一直在尝试使用datastax spark-cassandra连接器(https://github.com/datastax/spark-cassandra-connector)从csv文件导入一些数据.我知道大多数情况下可以在导入时使用案例类,但是我正在处理大约500个字段的行,因此我不能在没有嵌套的情况下使用它们(由于案例的22个字段限制).也可以直接存储地图,但我不认为这是理想的,因为有几种数据类型.

我可能在RDD [String] - > RDD [(String,String,...)]的转换中遗漏了一些东西,因为.split(",")只产生RDD [Array [String]].

我做了相当多的搜索没有太多运气,所以任何帮助将不胜感激!谢谢.

Answer 1

我会做这样的事情:

1. 读取您的文本文件(或任何文件格式)
2. 使用.map( ..)在每一行转换成Array[Any](或Map[String,Any])
3. 这里有两个选择
   • 将每个转换Array[Any]为CassandraRow.一个CassandraRow是刚刚columnNames:Array[String]和columnValues:Array[Any],然后写RDD[CassandraRow]
   • 实现RowWriterFactory[Array[Any]]并RDD[Array[Any]]使用自定义编写RowWriterFactory.看看CassandraRowWriter代码.