Spark with Cassandra输入/输出

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());
    }

}

数据处理器。主要方法：

// 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)，尽管此示例的文档量最小

我想知道Spark与Cassandra集成的最佳实践是什么。在我的实施中，最好的选择是什么？

Apache Spark 1.0.0，Apache Cassandra 2.0.8

Answer 1

上面代码中的方法是一种经典的集中式算法，只有在一个节点上执行才能工作。Cassandra和Spark都是分布式系统，因此有必要以一种可以分布在多个节点之间的方式对流程进行建模。

有几种可能的方法：如果您知道要获取的行的键，您可以执行如下简单的操作：（使用DataStaxJava驱动程序）

val data = sparkContext.parallelize(keys).map{key => 
   val cluster = val cluster = Cluster.builder.addContactPoint(host).build()
   val session  = cluster.connect(keyspace)
   val statement = session.prepare("...cql...);")
   val boundStatement = new BoundStatement(sttmt)
   session.execute(session.execute(boundStatement.bind(...data...)
}

这将有效地在Spark集群中分配密钥的获取。请注意如何在闭包内完成与C*的连接，因为这确保了在每个单独的分布式工作线程上执行任务时建立连接。

假设您的示例使用通配符（即密钥未知），那么使用Cassandra的Hadoop接口是一个不错的选择。问题中链接的Spark Cassandra示例说明了此Hadoop接口在Cassandra上的使用。

Callhtml" target="_blank">iope是一个库，它通过提供一个简单的API来访问该功能，从而封装了使用Hadoop接口的复杂性。它仅在Scala中可用，因为它使用特定的Scala功能（如即将发布的版本中的隐含和宏）使用Calliope，您基本上可以声明如何将RDD[type]转换为行键和行值，而Calliope负责将hadoop接口配置为作业。我们发现Calliope（和底层hadoop接口）比使用驱动程序与Cassandra交互快2-4倍。

结论：我将退出Spring数据配置来访问Cassandra，因为这将把您限制在单个节点上。如果可能的话，考虑一个简单的并行访问，或者在Scala中使用Calliope进行探索。

Answer 2

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

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

• 支持所有Cassandra数据类型，包括集合，开箱即用