Apache Spark技术实战之4 -- 利用Spark将json文件导入Cassandra
2014-09-06 14:54
375 查看
欢迎转载,转载请注明出处。
jdk
scala
sbt
cassandra
spark-cassandra-connector
但想到spark sql中可以读取json文件,而spark-cassadra-connector又提供了将RDD存入到数据库的功能,我想是否可以将两者结合一下。
注意:
jsonFile返回的是jsonRDD,其中每一个成员是Row类型,并不行直接将saveToCassandra作用于jsonRDD,需要先作一步转换即map过程
map中使用到的getXXX函数是在事先已知数据类型的情况下取出其值
最后saveToCassandra触发数据的存储过程
另外一个地方值得记录一下,如果在cassandra中创建的表使用了uuid作为primary key,在scala中使用如下函数来生成uuid
spark sql
spark RDD的转换函数
spark-cassandra-connector
概要
本文简要介绍如何使用spark-cassandra-connector将json文件导入到cassandra数据库,这是一个使用spark的综合性示例。前提条件
假设已经阅读技术实战之3,并安装了如下软件jdk
scala
sbt
cassandra
spark-cassandra-connector
实验目的
将存在于json文件中的数据导入到cassandra数据库,目前由cassandra提供的官方工具是json2sstable,由于对cassandra本身了解不多,这个我还没有尝试成功。但想到spark sql中可以读取json文件,而spark-cassadra-connector又提供了将RDD存入到数据库的功能,我想是否可以将两者结合一下。
创建KeySpace和Table
为了减少复杂性,继续使用实战3中的keyspace和table,CREATE KEYSPACE test WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1 }; CREATE TABLE test.kv(key text PRIMARY KEY, value int);
启动spark-shell
与实战3中描述一致。bin/spark-shell --driver-class-path /root/working/spark-cassandra-connector/spark-cassandra-connector/target/scala-2.10/spark-cassandra-connector_2.10-1.1.0-SNAPSHOT.jar:/root/.ivy2/cache/org.apache.cassandra/cassandra-thrift/jars/cassandra-thrift-2.0.9.jar:/root/.ivy2/cache/org.apache.thrift/libthrift/jars/libthrift-0.9.1.jar:/root/.ivy2/cache/org.apache.cassandra/cassandra-clientutil/jars/cassandra-clientutil-2.0.9.jar:/root/.ivy2/cache/com.datastax.cassandra/cassandra-driver-core/jars/cassandra-driver-core-2.0.4.jar:/root/.ivy2/cache/io.netty/netty/bundles/netty-3.9.0.Final.jar:/root/.ivy2/cache/com.codahale.metrics/metrics-core/bundles/metrics-core-3.0.2.jar:/root/.ivy2/cache/org.slf4j/slf4j-api/jars/slf4j-api-1.7.7.jar:/root/.ivy2/cache/org.apache.commons/commons-lang3/jars/commons-lang3-3.3.2.jar:/root/.ivy2/cache/org.joda/joda-convert/jars/joda-convert-1.2.jar:/root/.ivy2/cache/joda-time/joda-time/jars/joda-time-2.3.jar:/root/.ivy2/cache/org.apache.cassandra/cassandra-all/jars/cassandra-all-2.0.9.jar:/root/.ivy2/cache/org.slf4j/slf4j-log4j12/jars/slf4j-log4j12-1.7.2.jar
准备json文件
以spark自带的person.json文件为例,内容如下所示{"name":"Andy", "age":30} {"name":"Justin", "age":19}
数据导入
假设person.json文件存储在$SPARK_HOME目录,在启动spark-shell之后,执行如下语句sc.stop import com.datastax.spark.connector._ import org.apache.spark._ val conf = new SparkConf() conf.set("spark.cassandra.connection.host", "127.0.0.1") val sc = new SparkContext("local[2]", "Cassandra Connector Test", conf) val sqlContext = new org.apache.spark.sql.SQLContext(sc) val path = "./people.json" val people = sqlContext.jsonFile(path) people.map(p=>(p.getString(10),p.getInt(0))) .saveToCassandra("test","kv",SomeColumns("key","value"))
注意:
jsonFile返回的是jsonRDD,其中每一个成员是Row类型,并不行直接将saveToCassandra作用于jsonRDD,需要先作一步转换即map过程
map中使用到的getXXX函数是在事先已知数据类型的情况下取出其值
最后saveToCassandra触发数据的存储过程
另外一个地方值得记录一下,如果在cassandra中创建的表使用了uuid作为primary key,在scala中使用如下函数来生成uuid
import java.util.UUID UUID.randomUUID
验证步骤
使用cqlsh来查看数据是否已经真正的写入到test.kv表中。小结
本次实验结合了以下知识spark sql
spark RDD的转换函数
spark-cassandra-connector
相关文章推荐
- Apache Spark技术实战之8:Standalone部署模式下的临时文件清理
- Apache Spark技术实战之3 -- Spark Cassandra Connector的安装和使用
- Apache Spark技术实战之6 --Standalone部署模式下的临时文件清理
- Apache Spark技术实战之5 -- SparkR的安装及使用
- Apache Spark技术实战之6 -- spark-submit常见问题及其解决
- Apache Spark技术实战之6 -- spark-submit常见问题及其解决
- 利用 js-xlsx 实现 Excel 文件导入并解析Excel数据成json格式的数据并且获取其中某列数据
- Spark - 导入JSON文件 Error
- Apache Spark技术实战之2 -- PackratParsers实例
- Apache Spark技术实战之9 -- 日志级别修改
- Apache Spark技术实战之6 -- spark-submit常见问题及其解决
- Spark - 导入JSON、Text文件为Dataframe 做SQL查询
- CDH离线数据导入solr:利用MapReduceIndexerTool将json文件批量导入到solr
- Apache Spark技术实战之6 -- spark-submit常见问题及其解决
- Apache Spark技术实战之7 -- CassandraRDD高并发数据读取实现剖析
- 利用 js-xlsx 实现 Excel 文件导入并解析Excel数据成json格式的数据
- Apache Spark技术实战之1 -- KafkaWordCount
- 利用java的开源组件JExcel创建无差异的Excel文件,并且导入到.net项目中去.
- Flex中利用URLLoader和URLVariables类导入文件的例子
- 利用AJAX技术开发应用程序实战