spark如何高效的读写SequoiaDB的数据

使用更先进的浏览器访问效果更佳。

spark如何高效的读写SequoiaDB的数据

hustwei

比如对于传统读写hadoop数据方式：

sc.textFile(“hdfs://...”).flatMap(_.split(“ ”)).map(_, 1).reduceByKey(_ + _).map(x => (x._2, x._1)).sortByKey(false).map(x => (x._2, x_1)).saveAsTextFile(“hdfs://…")

实际执行在action中，因为数据在hadoop中分三份放，读取出来是充分并行的，直接在hadoop上进行。

但是SequoiaDB似乎没有类似textFile这样的方式，我想到的只能这样(java表达)：

List> list = new ArrayList>();

DBCursor cursor = cl.query(where, field, null, null);

while(cursor.hasNext()) {

BSONObject o = cursor.getNext();

String key = (String)o.get("Key");

String value = (String)o.get("Value");

list.add(new Tuple2(key, value));

}

JavaPairRDD inputRDD = sc.parallelizePairs(list );

result = inputRDD.mapPartitions(...).reduce(...);

这个准备inputRDD数据源的过程其实是在driver单进程进行，太浪费时间了，有没有好的解决方案。

wangzhonnew

spark + sdb是对每个数据分区中每个集合（如果是分区集合则是每个子集合）起一个任务，因此增加数据分区以及子集合的切分粒度能够更好滴增加并发性

wangzhonnew

https://github.com/SequoiaDB/spark-sequoiadb

可以参见spark-sequoiadb驱动的实现机制

hustwei

恩，这个库我看过，它是采用 Spark SQL读写SequoiaDB：

scala> sqlContext.sql("insert into table foo select * from jsontable")

但似乎不太满足我的需求（如果理解错了请指教）。

我想要的是类似spark-cassandra-connector-java 的形式：

import com.datastax.spark.connector.CassandraRow;

import static com.datastax.spark.connector.CassandraJavaUtil.javaFunctions;

// Read entire table as an RDD. Assumes your table test was created as

// CREATE TABLE test.kv(key text PRIMARY KEY, value int);

JavaRDD data = javaFunctions(sc).cassandraTable(“test” , “kv”);

// Print some basic stats.

System.out.println(data.mapToDouble(new DoubleFunction() {

public double call(CassandraRow row) { return row.getInt(“value”); }

}).stats());

从NoSQL的表直接映射成RDD，而实际数据的读取是在对RDD进行action操作后再执行。

hustwei

我又看了下spark从Hive中load数据的代码可以这样：

import org.apache.spark.sql.hive.HiveContext;

import org.apache.spark.sql.Row;

import org.apache.spark.sql.SchemaRDD;

HiveContext hiveCtx = new HiveContext(sc);

SchemaRDD rows = hiveCtx.sql(“SELECT key, value FROM mytable”);

JavaRDD keys = rdd.toJavaRDD().map(new Function() {

public Integer call(Row row) { return row.getInt(0); }

});

通过SchemaRDD的形式来实现，你们提供的spark-SequoiaDB-driver也可以采用sql方式，但怎么得到SchemaRDD呢？源码中也有SequoiadbRDD，或者得到这个也可以，制只要能直接进行后面的map/reduce都可以。

hustwei

王总，有解决方案吗？若没有，有没有计划开发呢？

我们还比较需要这样的功能，这样可以直接基于NoSQL提交spark计算，否则只能先把巨杉的数据写入到hdfs，再在hdfs基础上启动mapreduce或spark计算，并最后把结果存到巨杉了。