始める前に
このツールは、ファイルストリーミングインターフェイスを使用して、ApsaraDB for Cassandra クラスターにデータをインポートします。BulkLoad は、オフラインデータを Cassandra クラスターに移行する最速の方法の 1 つです。データをインポートする前に、次の準備を行ってください。
Cassandra クラスターを作成します。
SSTable または CSV 形式でオフラインデータを準備します。
Cassandra クラスターと同じ VPC に独立した ECS インスタンスを作成し、セキュリティグループルールを構成して、ECS インスタンスが Cassandra クラスターにアクセスできるようにします。
1. Cassandra クラスターと同じ VPC にクライアントの ECS インスタンスを作成する
Cassandra クラスターとは独立した ECS インスタンスを作成することをお勧めします。そうしないと、オンラインサービスが影響を受ける可能性があります。
2. スキーマを作成する
$ cqlsh -f schema.cql -u USERNAME -p PASSWORD [ホスト]
3. データを準備する
3.1 SSTable データ形式
data/${keyspace}/${table} 形式でディレクトリを編成し、次の例に示すように、SSTable データをディレクトリに保存します。
ls /tmp/quote/historical_prices/
md-1-big-CompressionInfo.db md-1-big-Data.db md-1-big-Digest.crc32 md-1-big-Filter.db md-1-big-Index.db md-1-big-Statistics.db md-1-big-Summary.db md-1-big-TOC.txt
前の例では、キースペースパラメーターは quote に設定され、テーブルパラメーターは historical_prices に設定されています。
データをインポートする
sstableloader コマンドを実行して、Cassandra ディストリビューションの bin ディレクトリにあるデータカタログ data/${ks}/${table} を指定します。
${cassandra_home}/bin/sstableloader -d <ノードの IP アドレス> data/${ks}/${table}
SSTable データをインポートした後、次のコマンドを実行してデータを確認します。 bin/cqlsh -u USERNAME -p PASSWORD [ホスト]
$ bin/cqlsh
cqlsh> select * from quote.historical_prices;
ticker | date | adj_close | close | high | low | open | volume
--------+---------------------------------+-----------+-----------+-----------+-----------+-----------+--------
ORCL | 2019-10-29 16:00:00.000000+0000 | 26.160000 | 26.160000 | 26.809999 | 25.629999 | 26.600000 | 181000
ORCL | 2019-10-28 16:00:00.000000+0000 | 26.559999 | 26.559999 | 26.700001 | 22.600000 | 22.900000 | 555000
3.2 CSV データ形式
最初に CSV データを SSTable 形式に変換する必要があります。Cassandra は、SSTable を生成するための CQLSSTableWriter ツールを提供しています。このツールを使用すると、任意の形式のデータを SSTable 形式に変換できます。CSV データも事前に整理する必要があります。したがって、CSV データを解析するためのコードを自分でコンパイルして実行する必要があります。次のサンプルコードは、このツールの使用方法を示しています。このツールの詳細については、GitHub リポジトリをご覧ください。
// SSTable writer を準備する
CQLSSTableWriter.Builder builder = CQLSSTableWriter.builder();
// 出力ディレクトリを設定する
builder.inDirectory(outputDir)
// ターゲットスキーマを設定する
.forTable(SCHEMA)
// データを配置するための CQL 文を設定する
.using(INSERT_STMT)
// 必要に応じて partitioner を設定する
// デフォルトは Murmur3Partitioner なので、異なるものを使用する場合は設定します。
.withPartitioner(new Murmur3Partitioner());
CQLSSTableWriter writer = builder.build();
/TODO: CSV ファイルを読み取ります。CSV ファイルの各行を反復的に読み取ります。
while ((line = csvReader.read()) ! = null)
{
writer.addRow(ticker,
DATE_FORMAT.parse(line.get(0)),
new BigDecimal(line.get(1)),
new BigDecimal(line.get(2)),
new BigDecimal(line.get(3)),
new BigDecimal(line.get(4)),
Long.parseLong(line.get(6)),
new BigDecimal(line.get(5)));
}
writer.close();
カスタムプログラムを使用して SSTable データを生成した後、セクション 3.1 の説明に従ってデータをインポートします。