Apache CassandraのNode.jsクライアント（DataStax Node.js Driver）を試す

Apache Cassandraには、いろいろな言語のドライバーが実装されています。

Client drivers

DataStaxドキュメント・ホーム

今回は、この中でNode.jsのクライアントを試してみたいと思います。

DataStax Nodejs Driver - Home

GitHub - datastax/nodejs-driver: DataStax Node.js Driver for Apache Cassandra

DataStaxの提供するドライバーですね。

インストール

まずは、インストール。npmプロジェクトの配下で、以下を実行。

$ npm i --save cassandra-driver

これで、DataStaxのNode.jsドライバーが利用可能になります。

Apache Cassandra側の準備

Apache Cassandraを起動しておき、キースペースを作成します。

cqlsh> CREATE KEYSPACE IF NOT EXISTS my_keyspace WITH REPLICATION = { 'class': 'SimpleStrategy', 'replication_factor': 3 };
cqlsh> USE my_keyspace;

テーブルも作成しておきましょう。お題は書籍とします。

cqlsh:my_keyspace> CREATE TABLE book ( isbn text PRIMARY KEY, title text, price int);

また、今回はApache Cassandraにリモートから接続します。

DataStax Node.js Driverを使う

それでは、DataStaxのNode.js Driverを使ってコードを書いていきましょう。

基本的な使い方は、こちらにそのまま書いてあるというのと

DataStax Node.js Driver for Apache Cassandra / Basic usage

データタイプや

DataStax Nodejs Driver - CQL data types to JavaScript types

APIリファレンスを見ていけばだいたいOKでしょう（？）。

DataStax Nodejs Driver - API docs

Cassandraに接続する

では、始めていきます。最初に「cassandra-driver」をrequireして、それからClientを作成します。接続先は、今回は
「172.17.0.2」とキースペースは先ほど作成した「my_keyspace」としました。

const cassandra = require("cassandra-driver");

const client = new cassandra.Client({ contactPoints: ["172.17.0.2"], keyspace: "my_keyspace" });

データを登録してみる

Node.js Driverのプログラムは、Promiseを使って書くかコールバック関数を定義するかのどちらかのスタイルで
書くことになります。

今回は、Promoiseを使いましょう。

では、データ登録。対象のデータは、こちらとします。

const books = [
    { isbn: "978-4873115290", title: "Cassandra", price: 3672 },
    { isbn: "978-4798128436", title: "Cassandra実用システムインテグレーション", price: 3456 },
    { isbn: "978-4798138824", title: "HBase徹底入門 Hadoopクラスタによる高速データベースの実現", price: 4298 }
];

それぞれ、登録してみます。

client
    .execute("INSERT INTO book (isbn, title, price) VALUES (:isbn, :title, :price)", books[0], { prepare: true })
    .then(() => client.execute("INSERT INTO book (isbn, title, price) VALUES (?, ?, ?)", [books[1].isbn, books[1].title, books[1].price], { prepare: true }))
    .then(() => client.execute("INSERT INTO book (isbn, title, price) VALUES (:isbn, :title, :price)", books[2], { prepare: true }))

CQLの発行には、executeメソッドを使用します。

実は、今回全部PreparedStatementを使っています。PreparedStatementの書き方には2種類あって、「:名前」と「?」の
2つの書き方があります。

DataStax Node.js Driver for Apache Cassandra / Prepare your queries

まずは名前で指定する方法。

client
    .execute("INSERT INTO book (isbn, title, price) VALUES (:isbn, :title, :price)", books[0], { prepare: true })

第1引数がCQL、第2引数にバインドするパラメータを渡すわけですが、この時オブジェクトを渡します。その時の
メンバーの名前がバインドするパラメーターの名前と合っていればOKです。今回は、こちらですね。

    { isbn: "978-4873115290", title: "Cassandra", price: 3672 },

また、最後にオプションを指定するのですが、PreparedStatemtとして使う場合は以下の設定が必要です。

{ prepare: true }

続いて、「?」…要するにインデックス指定でパラメーターをバインドする方法です。この場合、バインドする
パラメーターを「?」の順番に配列として渡してあげればOKです。

client.execute("INSERT INTO book (isbn, title, price) VALUES (?, ?, ?)", [books[1].isbn, books[1].title, books[1].price], { prepare: true })

データを取得してみる

続いて、登録したデータを取得してみます。

SELECTの場合も、Clientのexecuteメソッドを使用します。

    .then(() => client.execute("SELECT * FROM book"))
    .then(result => {
        console.log(result);

        result.rows.forEach(row => {
            console.log(">>  isbn = " + row.isbn + ", title = " + row.title + ", price = " + row.price);
        });
    })

Promise経由で結果が渡ってくるので、今回はそれぞれconsole.logで出力してみました。

今回の例だと、以下のコードからは

        console.log(result);

こういう結果が出力され、

ResultSet {
  info: 
   { queriedHost: '172.17.0.2:9042',
     triedHosts: {},
     achievedConsistency: 10,
     traceId: undefined,
     warnings: undefined,
     customPayload: undefined },
  rows: 
   [ Row {
       isbn: '978-4798128436',
       price: 3456,
       title: 'Cassandra実用システムインテグレーション' },
     Row {
       isbn: '978-4798138824',
       price: 4298,
       title: 'HBase徹底入門 Hadoopクラスタによる高速データベースの実現' },
     Row { isbn: '978-4873115290', price: 3672, title: 'Cassandra' } ],
  rowLength: 3,
  columns: 
   [ { name: 'isbn', type: [Object] },
     { name: 'price', type: [Object] },
     { name: 'title', type: [Object] } ],
  pageState: null,
  nextPage: undefined }

こちらのコードからは

    .then(result => {
        console.log(result);

        result.rows.forEach(row => {
            console.log(">>  isbn = " + row.isbn + ", title = " + row.title + ", price = " + row.price);
        });
    })

こういう結果が得られます。

>>  isbn = 978-4798128436, title = Cassandra実用システムインテグレーション, price = 3456
>>  isbn = 978-4798138824, title = HBase徹底入門 Hadoopクラスタによる高速データベースの実現, price = 4298
>>  isbn = 978-4873115290, title = Cassandra, price = 3672

ちゃんとデータが取得できていますね。PreparedStatementは使っていませんが、INSERTで使った時と同じ書き方でOKです。

この他、eachRowやstreamといったメソッドもあるようなので、興味のある方は見てみるとよいでしょう。

DataStax Node.js Driver for Apache Cassandra / Row streaming and pipes

データを削除する

登録したデータを削除します。

削除では、バッチ更新をしてみました。

    .then(() => {
        const deleteCql = "DELETE FROM book WHERE isbn = :isbn";

        const queries = books.map((b) => {
            return {
                query: deleteCql,
                params: b
            }});

        return client.batch(queries, { prepare: true });
    })

バッチ更新には、Clientのbatchメソッドを使用します。

DataStax Node.js Driver for Apache Cassandra / Batch multiple statements

batchメソッドに、「query」と「params」からなる配列を渡してあげればOKです。また今回はPreparedStatementを
使っているので、「prepare: true」も指定しています。

切断

最後は、Cassandraからの切断。

    .then(() => client.shutdown());

Clientのshutdownメソッドを呼び出します。

まとめ

Apache CassandraにDataStax Node.js Driverを使ってアクセスしてみました。

自分がJavaScript、Node.jsに不慣れなのですが、ちょっとずつ慣れていきましょう。APIもそれほど難しいわけではないみたいですし。

最後に、今回作成したコード全体を載せておきます。
src/getting-started.js

const cassandra = require("cassandra-driver");

const client = new cassandra.Client({ contactPoints: ["172.17.0.2"], keyspace: "my_keyspace" });

const books = [
    { isbn: "978-4873115290", title: "Cassandra", price: 3672 },
    { isbn: "978-4798128436", title: "Cassandra実用システムインテグレーション", price: 3456 },
    { isbn: "978-4798138824", title: "HBase徹底入門 Hadoopクラスタによる高速データベースの実現", price: 4298 }
];

client
    .execute("INSERT INTO book (isbn, title, price) VALUES (:isbn, :title, :price)", books[0], { prepare: true })
    .then(() => client.execute("INSERT INTO book (isbn, title, price) VALUES (?, ?, ?)", [books[1].isbn, books[1].title, books[1].price], { prepare: true }))
    .then(() => client.execute("INSERT INTO book (isbn, title, price) VALUES (:isbn, :title, :price)", books[2], { prepare: true }))
    .then(() => client.execute("SELECT * FROM book"))
    .then(result => {
        console.log(result);

        result.rows.forEach(row => {
            console.log(">>  isbn = " + row.isbn + ", title = " + row.title + ", price = " + row.price);
        });
    })
    .then(() => {
        const deleteCql = "DELETE FROM book WHERE isbn = :isbn";

        const queries = books.map((b) => {
            return {
                query: deleteCql,
                params: b
            }});

        return client.batch(queries, { prepare: true });
    })
    .catch(e => console.log(e))
    .then(() => client.shutdown());