首页 文章

Spark与Cassandra输入/输出

提问于
浏览 次
8

想象一下以下场景：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中进行分页，加载和分发 .

还有以下选项：

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

Apache Spark 1.0.0，Apache Cassandra 2.0.8

java cassandra apache-spark spring-data-cassandra

2 回答

  • 9

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

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

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

    • 将Cassandra行轻量映射到自定义类或元组，而无需在Scala中使用任何implicits或其他高级功能

    • 将任何RDD保存到Cassandra

    • 完全支持Cassandra虚拟节点

    • 在服务器端过滤/选择的能力，例如利用Cassandra集群列或二级索引

    回复于
  • 1

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

    可能的方法很少：如果您知道要获取的行的键，您可以执行以下简单操作:(使用DataStax Java驱动程序）

    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 *的连接，因为这可确保在每个单独的分布式工作程序上执行任务时 Build 连接 .

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

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

    结论：我将离开Spring-Data配置来访问Cassandra，因为这将限制您到单个节点 . 如果可能，请考虑使用简单的并行访问，或者在Scala中使用Calliope进行探索 .

    回复于

相关问题