Spring Boot与Spark、Cassandra系统集成开发示例

CREATE KEYSPACE hfcb WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 3 };

CREATE TABLE person ( id text PRIMARY KEY, first_name text, last_name text);

insert into person (id,first_name,last_name) values('1','wang','yunfei');insert into person (id,first_name,last_name) values('2','peng','chao');insert into person (id,first_name,last_name) values('3','li','jian');insert into person (id,first_name,last_name) values('4','zhang','jie');insert into person (id,first_name,last_name) values('5','liang','wei');

cassandra-clientutil-3.0.2.jarcassandra-driver-core-3.1.4.jarguava-16.0.1.jarcassandra-thrift-3.0.2.jar joda-convert-1.2.jarjoda-time-2.9.9.jarlibthrift-0.9.1.jarspark-cassandra-connector_2.10-1.5.1.jar

SPARK_CLASSPATH=/opt/spark/managed-lib/*

<dependency>   <groupId>com.datastax.spark</groupId>   <artifactId>spark-cassandra-connector_2.10</artifactId>   <version>1.5.1</version>  </dependency>  <dependency>   <groupId>org.apache.spark</groupId>   <artifactId>spark-core_2.10</artifactId>   <version>1.5.1</version>  </dependency>  <dependency>   <groupId>org.apache.spark</groupId>   <artifactId>spark-sql_2.10</artifactId>   <version>1.5.1</version>  </dependency>

spark.master: spark://master:7077cassandra.host: 192.168.1.140cassandra.keyspace: hfcb

@Configurationpublic class SparkCassandraConfig { @Value("${spark.master}") String sparkMasterUrl; @Value("${cassandra.host}") String cassandraHost; @Value("${cassandra.keyspace}") String cassandraKeyspace; @Bean public JavaSparkContext javaSparkContext(){  SparkConf conf = new SparkConf(true)    .set("spark.cassandra.connection.host", cassandraHost)//    .set("spark.cassandra.auth.username", "cassandra")//    .set("spark.cassandra.auth.password", "cassandra")    .set("spark.submit.deployMode", "client");  JavaSparkContext context = new JavaSparkContext(sparkMasterUrl, "SparkDemo", conf);  return context; } @Bean public CassandraSQLContext sqlContext(){  CassandraSQLContext cassandraSQLContext = new CassandraSQLContext(javaSparkContext().sc());  cassandraSQLContext.setKeyspace(cassandraKeyspace);  return cassandraSQLContext; } }

@Repositorypublic class PersonRepository { @Autowired CassandraSQLContext cassandraSQLContext; public Long countPerson(){  DataFrame people = cassandraSQLContext.sql("select * from person order by id");  return people.count(); }}