Spring BootとSpark、Cassandraシステム統合開発例

解析エンジンとしてSparkを用いて，Cassandraをデータストアとして，Spring Bootを用いて駆動プログラムを開発する例を示した。
1.事前条件

はSparkをインストールします。（ここではSpark-1.5.1を使用して、インストールディレクトリが/opt/sparkのようです。）

Cassandra（3.0+）

をインストールします。
keyspaceを作成します

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');

2.spark-cassandra-connectorの設置
Spark-1.5.1をデータとしてCassandraを使用することができるようにするには、次のjarパケットの依存性が必要です。

cassandra-clientutil-3.0.2.jar
cassandra-driver-core-3.1.4.jar
guava-16.0.1.jar
cassandra-thrift-3.0.2.jar 
joda-convert-1.2.jar
joda-time-2.9.9.jar
libthrift-0.9.1.jar
spark-cassandra-connector_2.10-1.5.1.jar

opt/spark/confディレクトリの下で、spark-env.shファイルを新規作成して、下記の内容を入力します。

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

3.Spring Bootアプリケーション開発
spark-cassandra-connectorとspark依存を追加します。

<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>

appication.ymlにsparkとcassandra経路を配置する。

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

ここで特に説明しますspark://master:7077 IPアドレスではなくドメイン名です。ローカルのhostsファイルを修正して、masterとipアドレスをマッピングします。
SparkContectとCassandraSQLContextを設定します。

@Configuration
public 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;
 }
 }

簡単な呼び出し

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

起動は通常のSpring Bootプログラムのように実行できます。
ソースの住所：https://github.com/wiselyman/spring-spark-cassandra.git
締め括りをつける
以上は小编が绍介したSpring BootとSpark、Cassandraシステム统合开発例です。皆様に何かご迷惑があったら、メッセージをください。小编はすぐにご返事します。ここでも私たちのサイトを応援してくれてありがとうございます。