带Cassandra输入/输出的Spark

Question

请看下面的场景：一个Spark应用程序（Java实现）正在使用Cassandra数据库加载，转换为RDD并处理数据。该应用程序还从数据库中提取新数据，这些新数据也由自定义接收器处理。流处理的输出存储在数据库中。该实现使用了与数据库集成中的Spring
Data Cassandra。

CassandraConfig：

@Configuration
@ComponentScan(basePackages = {"org.foo"})
@PropertySource(value = { "classpath:cassandra.properties" })
public class CassandraConfig {

    @Autowired
    private Environment env;

    @Bean
    public CassandraClusterFactoryBean cluster() {
        CassandraClusterFactoryBean cluster = new CassandraClusterFactoryBean();
        cluster.setContactPoints(env.getProperty("cassandra.contactpoints"));
        cluster.setPort(Integer.parseInt(env.getProperty("cassandra.port")));

        return cluster;
    }

    @Bean
    public CassandraMappingContext mappingContext() {
        return new BasicCassandraMappingContext();
    }

    @Bean
    public CassandraConverter converter() {
        return new MappingCassandraConverter(mappingContext());
    }

    @Bean
    public CassandraSessionFactoryBean session() throws Exception {
        CassandraSessionFactoryBean session = new CassandraSessionFactoryBean();
        session.setCluster(cluster().getObject());
        session.setKeyspaceName(env.getProperty("cassandra.keyspace"));
        session.setConverter(converter());
        session.setSchemaAction(SchemaAction.NONE);

        return session;
    }

    @Bean
    public CassandraOperations cassandraTemplate() throws Exception {
        return new CassandraTemplate(session().getObject());
    }

}

DataProcessor.main方法：

// Initialize spring application context
ApplicationContext applicationContext = new AnnotationConfigApplicationContext(CassandraConfig.class);
ApplicationContextHolder.setApplicationContext(applicationContext);
CassandraOperations cassandraOperations = applicationContext.getBean(CassandraOperations.class);
// Initialize spark context
SparkConf conf = new SparkConf().setAppName("test-spark").setMaster("local[2]");
JavaSparkContext sc = new JavaSparkContext(conf);

// Load data pages
List<Event> pagingResults = cassandraOperations.select("select * from event where event_type = 'event_type1' order by creation_time desc limit " + DATA_PAGE_SIZE, Event.class);
// Parallelize the first page
JavaRDD<Event> rddBuffer = sc.parallelize(pagingResults);

while(pagingResults != null && !pagingResults.isEmpty()) {
    Event lastEvent = pagingResults.get(pagingResults.size() - 1);
    pagingResults = cassandraOperations.select("select * from event where event_type = 'event_type1' and creation_time < " + lastEvent.getPk().getCreationTime() + " order by creation_time desc limit " + DATA_PAGE_SIZE, Event.class);
    // Parallelize page and add to the existing
    rddBuffer = rddBuffer.union(sc.parallelize(pagingResults));
}

// data processing
...

预计在初始加载时会有大量数据。因此，将数据分页，装入并分发到rddBuffer中。

还有以下可用选项：

1. 尽管文档数量很少，但Spark-Cassandra示例（https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/CassandraCQLTest.scala）对于这个例子。
2. Calliope项目（http://tuplejump.github.io/calliope/）

我想知道将Spark与Cassandra集成的最佳实践是什么。在我的实施过程中遵循的最佳选择是什么？

Apache Spark 1.0.0，Apache Cassandra 2.0.8

Answer 1

与Cassandra和Spark一起使用的最简单方法是使用由DataStax开发的Spark的官方开源Cassandra驱动程序：https : //github.com/datastax/spark-cassandra-
connector

该驱动程序基于Cassandra
Java驱动程序构建，并提供了Cassandra和Spark之间的直接桥梁。与Calliope不同，它不使用Hadoop接口。此外，它还提供以下独特功能：

• 开箱即用地支持所有Cassandra数据类型，包括集合
• Cassandra行到自定义类或元组的轻量级映射，而无需在Scala中使用任何隐式或其他高级功能
• 将所有RDD保存到Cassandra
• 完全支持Cassandra虚拟节点
• 在服务器端进行过滤/选择的能力，例如，利用Cassandra集群列或二级索引