Accumuloコネクタ¶

概要¶

Accumuloコネクタは、Apache Accumuloからのデータの読み取りと書き込みをサポートしています。コネクタの機能と特徴を理解するために、このページをよく読んでください。

イテレータ依存関係のインストール¶

Accumuloコネクタは、*述語プッシュダウン*として知られるサーバーサイドフィルタリングのために、SQL述語句の様々な情報をAccumuloにプッシュするために、カスタムAccumuloイテレータを使用します。サーバーサイドイテレータが動作するためには、各TabletServerノードのAccumuloの`lib/ext`ディレクトリに`presto-accumulo` jarファイルを追加する必要があります。

# For each TabletServer node:
scp $PRESTO_HOME/plugins/accumulo/presto-accumulo-*.jar [tabletserver_address]:$ACCUMULO_HOME/lib/ext

# TabletServer should pick up new JAR files in ext directory, but may require restart

これはJava 8を使用することに注意してください。AccumuloクラスタがJava 7を使用している場合、インデックス付きテーブルを作成しようとすると、TabletServerログに`Unsupported major.minor version 52.0`エラーが表示されます。代わりに、https://github.com/bloomberg/presto-accumuloにある*presto-accumulo-iterators* jarファイルを使用する必要があります。

コネクタの設定¶

`accumulo.xxx`プロパティを必要に応じて置き換え、`accumulo`コネクタを`accumulo`カタログとしてマウントするために、`etc/catalog/accumulo.properties`を作成します。

connector.name=accumulo
accumulo.instance=xxx
accumulo.zookeepers=xxx
accumulo.username=username
accumulo.password=password

設定変数¶

サポートされていない機能¶

以下の機能はサポートされていません

• `ALTER TABLE`によるカラムの追加: SQLでカラムを追加することはできませんが、ツールを使用すれば可能です。詳細については、以下のカラムの追加セクションを参照してください。

• `DELETE`: 行の削除はまだコネクタに実装されていません。

使用方法¶

データを扱うには、単純にSQLを使用してAccumuloに新しいテーブルを作成することから始めます。デフォルトでは、テーブル定義の最初のカラムはAccumuloの行IDに設定されます。これはテーブルのプライマリキーである必要があります。また、同じ行IDを含む`INSERT`文は、セル内の以前のデータが上書きされるため、Accumuloに関しては事実上UPDATEであることに注意してください。行IDは、有効なPrestoデータ型であれば何でも構いません。最初のカラムがプライマリキーでない場合は、テーブル定義の`WITH`句内で`row_id`テーブルプロパティを使用して行IDカラムを設定できます。

新しいPresto/Accumuloテーブルを作成するには、単純に`CREATE TABLE`ステートメントを実行します。

CREATE TABLE myschema.scientists (
  recordkey VARCHAR,
  name VARCHAR,
  age BIGINT,
  birthday DATE
);

DESCRIBE myschema.scientists;

  Column   |  Type   | Extra |                      Comment
-----------+---------+-------+---------------------------------------------------
 recordkey | varchar |       | Accumulo row ID
 name      | varchar |       | Accumulo column name:name. Indexed: false
 age       | bigint  |       | Accumulo column age:age. Indexed: false
 birthday  | date    |       | Accumulo column birthday:birthday. Indexed: false

このコマンドは、`recordkey`カラムをAccumulo行IDとして使用して、新しいAccumuloテーブルを作成します。name、age、birthdayカラムは、自動生成されたカラムファミリと修飾子値にマッピングされます（実際には、どちらもPrestoカラム名と同じです）。

SQLを使用してテーブルを作成する場合、オプションで`column_mapping`テーブルプロパティを指定できます。このプロパティの値は、prestoカラム**:** accumuloカラムファミリ**:** accumuloカラム修飾子の3つ組をカンマ区切りでリストしたもので、行ID以外のカラムごとに1つの3つ組が必要です。これは、Prestoカラム名を対応するAccumuloカラムファミリとカラム修飾子にマッピングします。

`column_mapping`テーブルプロパティを指定しない場合、コネクタはカラム名を自動生成します（設定されたローカリティグループを尊重します）。カラム名の自動生成は内部テーブルでのみ使用できるため、テーブルが外部テーブルの場合はcolumn_mappingプロパティを指定する必要があります。

テーブルプロパティの完全なリストについては、テーブルプロパティを参照してください。

例えば

CREATE TABLE myschema.scientists (
  recordkey VARCHAR,
  name VARCHAR,
  age BIGINT,
  birthday DATE
)
WITH (
  column_mapping = 'name:metadata:name,age:metadata:age,birthday:metadata:date'
);

DESCRIBE myschema.scientists;

  Column   |  Type   | Extra |                    Comment
-----------+---------+-------+-----------------------------------------------
 recordkey | varchar |       | Accumulo row ID
 name      | varchar |       | Accumulo column metadata:name. Indexed: false
 age       | bigint  |       | Accumulo column metadata:age. Indexed: false
 birthday  | date    |       | Accumulo column metadata:date. Indexed: false

その後、`INSERT`ステートメントを実行して、Accumuloにデータを挿入できます。

INSERT INTO myschema.scientists VALUES
('row1', 'Grace Hopper', 109, DATE '1906-12-09' ),
('row2', 'Alan Turing', 103, DATE '1912-06-23' );

SELECT * FROM myschema.scientists;

 recordkey |     name     | age |  birthday
-----------+--------------+-----+------------
 row1      | Grace Hopper | 109 | 1906-12-09
 row2      | Alan Turing  | 103 | 1912-06-23
(2 rows)

ご想像のとおり、シェルまたはプログラムでAccumuloに挿入された行も、クエリを実行すると表示されます。(Accumuloシェルは「-5321」をオプションとみなし、数値とみなさないため、TBLを少し若くします)。

$ accumulo shell -u root -p secret
root@default> table myschema.scientists
root@default myschema.scientists> insert row3 metadata name "Tim Berners-Lee"
root@default myschema.scientists> insert row3 metadata age 60
root@default myschema.scientists> insert row3 metadata date 5321

SELECT * FROM myschema.scientists;

 recordkey |      name       | age |  birthday
-----------+-----------------+-----+------------
 row1      | Grace Hopper    | 109 | 1906-12-09
 row2      | Alan Turing     | 103 | 1912-06-23
 row3      | Tim Berners-Lee |  60 | 1984-07-27
(3 rows)

`DROP TABLE`を使用してテーブルを削除することもできます。このコマンドは、メタデータとテーブルの両方を削除します。内部テーブルと外部テーブルの詳細については、以下の外部テーブルセクションを参照してください。

DROP TABLE myschema.scientists;

カラムのインデックス化¶

内部的には、コネクタは Accumulo の Range を作成し、それをスプリットにパックします。このスプリットは Presto ワーカーに渡され、BatchScanner を介して Range からデータを読み取ります。フルテーブルスキャンとなるクエリを発行すると、各 Presto ワーカーはテーブルの単一のタブレットに対応する単一の Range を取得します。述語（つまり、WHERE x = 10 句）を含むクエリを発行する場合、Presto は述語内の値（10）をコネクタに渡し、コネクタはこの情報を使用してスキャンするデータ量を削減できます。Accumulo 行 ID が述語句の一部として使用される場合、これは Range ルックアップを絞り込み、Accumulo からデータのサブセットを迅速に取得します。

しかし、他のカラムはどうでしょうか？行 ID 以外のカラムを頻繁にクエリする場合、Accumulo コネクタに組み込まれている **インデックス** 機能の使用を検討する必要があります。この機能は、テーブルから少数の値を選択する際のクエリ実行時間を大幅に短縮でき、Presto の INSERT 文を介してデータを読み込む際に面倒な作業が自動的に行われます（ただし、INSERT を介して Accumulo にデータを書き込む場合、スループットは高くないことに注意してください）。

インデックスを有効にするには、index_columns テーブルプロパティを追加し、インデックスを作成する Presto カラム名のカンマ区切りリストを指定します（ここでは、この例を分かりやすくするために string シリアライザを使用しています。デフォルトの lexicoder シリアライザを使用する必要があります）。

CREATE TABLE myschema.scientists (
  recordkey VARCHAR,
  name VARCHAR,
  age BIGINT,
  birthday DATE
)
WITH (
  serializer = 'string',
  index_columns='name,age,birthday'
);

テーブルを作成後、インデックスとメトリクスを格納するための追加の Accumulo テーブルが 2 つあることがわかります。

root@default> tables
accumulo.metadata
accumulo.root
myschema.scientists
myschema.scientists_idx
myschema.scientists_idx_metrics
trace

データを挿入した後、インデックステーブルを見ると、名前、年齢、誕生日カラムのインデックス付き値があることがわかります。コネクタはこのインデックステーブルをクエリします

INSERT INTO myschema.scientists VALUES
('row1', 'Grace Hopper', 109, DATE '1906-12-09'),
('row2', 'Alan Turing', 103, DATE '1912-06-23');

root@default> scan -t myschema.scientists_idx
-21011 metadata_date:row2 []
-23034 metadata_date:row1 []
103 metadata_age:row2 []
109 metadata_age:row1 []
Alan Turing metadata_name:row2 []
Grace Hopper metadata_name:row1 []

インデックス付きカラムに対して WHERE 句を含むクエリを発行すると、コネクタは述語内の値を含むすべての行 ID についてインデックステーブルを検索します。これらの行 ID は、単一値の Range オブジェクトとして Presto スプリットにバンドルされ（スプリットごとの行 ID の数は accumulo.index_rows_per_split の値によって制御されます）、Presto ワーカーに渡されてデータテーブルをスキャンする BatchScanner で構成されます。

SELECT * FROM myschema.scientists WHERE age = 109;

 recordkey |     name     | age |  birthday
-----------+--------------+-----+------------
 row1      | Grace Hopper | 109 | 1906-12-09
(1 row)

データのロード¶

Accumulo コネクタは INSERT 文によるデータのロードをサポートしていますが、この方法はスループットが低くなる傾向があり、スループットが重要な場合は依存すべきではありません。代わりに、コネクタのユーザーは、presto-accumulo リポジトリ の presto-accumulo-tools サブプロジェクトの一部として提供されている PrestoBatchWriter ツールを使用する必要があります。

PrestoBatchWriter は、Presto/Accumulo メタデータを利用してメインデータテーブルに Mutations を書き込む、一般的な BatchWriter のラッパークラスです。特に、インデックス付きの任意のカラムで指定されたミューテーションのインデックスを作成します。ツールの使い方は、リポジトリ の README に記載されています。

外部テーブル¶

デフォルトでは、Presto を介して SQL 文を使用して作成されたテーブルは *内部* テーブルです。つまり、Presto テーブルメタデータと Accumulo テーブルの両方が Presto によって管理されます。内部テーブルを作成すると、Accumulo テーブルも作成されます。Accumulo テーブルが既に存在する場合、エラーが発生します。Presto を介して内部テーブルが削除されると、Accumulo テーブル（およびインデックステーブル）も削除されます。

この動作を変更するには、CREATE 文を発行するときに external プロパティを true に設定します。これにより、テーブルは *外部* テーブルになり、DROP TABLE コマンドはテーブルに関連付けられたメタデータ **のみ** を削除します。Accumulo テーブルがまだ存在しない場合、コネクタによって作成されます。

外部テーブルを作成すると、構成済みのローカリティグループと、インデックスおよびメトリックステーブルのイテレータが設定されます（テーブルにインデックスが付けられている場合）。つまり、外部テーブルと内部テーブルの唯一の違いは、DROP TABLE コマンドが発行されたときにコネクタが Accumulo テーブルを削除することです。

外部テーブルは、データが期待される形式で格納されるため、操作が少し難しい場合があります。データが正しく格納されていない場合、問題が発生します。ユーザーは、テーブルを作成するときに column_mapping プロパティを提供する必要があります。これにより、Presto カラム名とテーブルのセルのカラムファミリ/クオリファイアの対応付けが作成されます。セルの値は、Accumulo キー/値ペアの Value に格納されます。デフォルトでは、この値は Accumulo の *lexicoder* API を使用してシリアル化されることが想定されています。値を文字列として格納している場合は、テーブルの serializer プロパティを使用して別のシリアライザを指定できます。詳細については、テーブルプロパティ のセクションを参照してください。

次に、Presto 外部テーブルを作成します。

CREATE TABLE external_table (
  a VARCHAR,
  b BIGINT,
  c DATE
)
WITH (
  column_mapping = 'a:md:a,b:md:b,c:md:c',
  external = true,
  index_columns = 'b,c',
  locality_groups = 'foo:b,c'
);

テーブルを作成した後、テーブルの使い方は通常どおりです

INSERT INTO external_table VALUES
('1', 1, DATE '2015-03-06'),
('2', 2, DATE '2015-03-07');

SELECT * FROM external_table;

 a | b |     c
---+---+------------
 1 | 1 | 2015-03-06
 2 | 2 | 2015-03-06
(2 rows)

DROP TABLE external_table;

テーブルを削除した後も、テーブルは *外部* であるため、Accumulo に存在し続けます。

root@default> tables
accumulo.metadata
accumulo.root
external_table
external_table_idx
external_table_idx_metrics
trace

テーブルに新しいカラムを追加する場合、テーブルを再度作成し、新しいカラムを指定できます。テーブルの既存の行の値は NULL になります。このコマンドは、Accumulo テーブルを再構成し、ローカリティグループとイテレータ構成を設定します。

CREATE TABLE external_table (
  a VARCHAR,
  b BIGINT,
  c DATE,
  d INTEGER
)
WITH (
  column_mapping = 'a:md:a,b:md:b,c:md:c,d:md:d',
  external = true,
  index_columns = 'b,c,d',
  locality_groups = 'foo:b,c,d'
);

SELECT * FROM external_table;

 a | b |     c      |  d
---+---+------------+------
 1 | 1 | 2015-03-06 | NULL
 2 | 2 | 2015-03-07 | NULL
(2 rows)

テーブルプロパティ¶

テーブルプロパティの使用例

CREATE TABLE myschema.scientists (
  recordkey VARCHAR,
  name VARCHAR,
  age BIGINT,
  birthday DATE
)
WITH (
  column_mapping = 'name:metadata:name,age:metadata:age,birthday:metadata:date',
  index_columns = 'name,age'
);

セッションプロパティ¶

SET SESSION を使用して、セッションプロパティのデフォルト値を変更できます。セッションプロパティにはカタログ名のプレフィックスが付いていることに注意してください

SET SESSION accumulo.column_filter_optimizations_enabled = false;

カラムの追加¶

カラムが機能するために必要な追加のメタデータ（カラムファミリ、クオリファイア、カラムにインデックスが付けられているかどうか）があるため、ALTER TABLE [table] ADD COLUMN [name] [type] を使用して既存のテーブルに新しいカラムを追加することはできません。

代わりに、presto-accumulo リポジトリの presto-accumulo-tools サブプロジェクトにあるユーティリティのいずれかを使用できます。ドキュメントと使い方は README にあります。

シリアライザ¶

Accumulo 用の Presto コネクタには、Presto と Accumulo 間の I/O を処理するためのプラグイン可能なシリアライザフレームワークがあります。これにより、エンドユーザーは Accumulo 内で特別なデータ形式をプログラムでシリアル化および逆シリアル化でき、コネクタ自体の複雑さを抽象化できます。

現在、2種類のシリアライザーが利用可能です。値をJavaのStringとして扱うstringシリアライザーと、AccumuloのLexicoder APIを活用して値を格納するlexicoderシリアライザーです。デフォルトのシリアライザーはlexicoderシリアライザーです。これは、StringオブジェクトとPrestoの型の間で高コストな変換操作を必要としないためです。セルの値はバイト配列としてエンコードされます。

さらに、lexicoderシリアライザーは、BIGINTやTIMESTAMPのような数値型を適切な辞書順に並べ替えます。これは、コネクタがデータのクエリ時にセカンダリインデックスを適切に活用するために不可欠です。

デフォルトのシリアライザーは、serializerテーブルプロパティで変更できます。default（lexicoderと同じ）、string、またはlexicoderを組み込み型として使用するか、AccumuloRowSerializerを拡張して独自の実装を提供し、PrestoのCLASSPATHに追加し、コネクタ設定で完全修飾Javaクラス名を指定することもできます。

CREATE TABLE myschema.scientists (
  recordkey VARCHAR,
  name VARCHAR,
  age BIGINT,
  birthday DATE
)
WITH (
  column_mapping = 'name:metadata:name,age:metadata:age,birthday:metadata:date',
  serializer = 'default'
);

INSERT INTO myschema.scientists VALUES
('row1', 'Grace Hopper', 109, DATE '1906-12-09' ),
('row2', 'Alan Turing', 103, DATE '1912-06-23' );

root@default> scan -t myschema.scientists
row1 metadata:age []    \x08\x80\x00\x00\x00\x00\x00\x00m
row1 metadata:date []    \x08\x7F\xFF\xFF\xFF\xFF\xFF\xA6\x06
row1 metadata:name []    Grace Hopper
row2 metadata:age []    \x08\x80\x00\x00\x00\x00\x00\x00g
row2 metadata:date []    \x08\x7F\xFF\xFF\xFF\xFF\xFF\xAD\xED
row2 metadata:name []    Alan Turing

CREATE TABLE myschema.stringy_scientists (
  recordkey VARCHAR,
  name VARCHAR,
  age BIGINT,
  birthday DATE
)
WITH (
  column_mapping = 'name:metadata:name,age:metadata:age,birthday:metadata:date',
  serializer = 'string'
);

INSERT INTO myschema.stringy_scientists VALUES
('row1', 'Grace Hopper', 109, DATE '1906-12-09' ),
('row2', 'Alan Turing', 103, DATE '1912-06-23' );

root@default> scan -t myschema.stringy_scientists
row1 metadata:age []    109
row1 metadata:date []    -23034
row1 metadata:name []    Grace Hopper
row2 metadata:age []    103
row2 metadata:date []    -21011
row2 metadata:name []    Alan Turing

CREATE TABLE myschema.custom_scientists (
  recordkey VARCHAR,
  name VARCHAR,
  age BIGINT,
  birthday DATE
)
WITH (
  column_mapping = 'name:metadata:name,age:metadata:age,birthday:metadata:date',
  serializer = 'my.serializer.package.MySerializer'
);

メタデータ管理¶

Presto/AccumuloテーブルのメタデータはZooKeeperに格納されます。PrestoでSQLステートメントを発行して、テーブルの作成と削除を行うことができます（そして、行うべきです）。これは、コネクタを動作させるために必要なメタデータを作成する最も簡単な方法です。メタデータを変更しないのが最善ですが、メタデータの格納方法の詳細を以下に示します。情報は力です。

ZooKeeperのルートノードはすべてのマッピングを保持し、フォーマットは以下のとおりです。

/metadata-root/schema/table

metadata-rootは設定ファイルのzookeeper.metadata.rootの値（デフォルトは/presto-accumulo）、schemaはPrestoスキーマ（Accumuloの名前空間名と同じ）、tableはPrestoテーブル名（これもAccumuloの名前と同じ）です。table ZooKeeperノードのデータは、シリアライズされたAccumuloTable Javaオブジェクト（コネクタコード内に存在）です。このテーブルには、スキーマ（名前空間）名、テーブル名、列定義、テーブルに使用するシリアライザー、および追加のテーブルプロパティが含まれています。

AccumuloのZooKeeperメタデータをプログラムで操作する必要がある場合は、プロセスを簡素化するJavaコードについてcom.facebook.presto.accumulo.metadata.ZooKeeperMetadataManagerを参照してください。

内部テーブルから外部テーブルへの変換¶

テーブルが*内部*テーブルの場合、ZooKeeperの対応するznodeを削除することで外部テーブルに変換できます。これにより、Prestoから見るとテーブルは事実上存在しなくなります。その後、同じDDLを使用してテーブルを再作成しますが、external = trueテーブルプロパティを追加します。

例えば

1. 以下のDDLで作成された内部テーブルfoo.barから始めます。column_mappingのテーブルプロパティを（この例のように）以前に定義していない場合は、メタデータを削除する**前に**テーブルを記述してください。外部テーブルを作成するときに列マッピングが必要になります。

CREATE TABLE foo.bar (a VARCHAR, b BIGINT, c DATE)
WITH (
    index_columns = 'b,c'
);

DESCRIBE foo.bar;

 Column |  Type   | Extra |               Comment
--------+---------+-------+-------------------------------------
 a      | varchar |       | Accumulo row ID
 b      | bigint  |       | Accumulo column b:b. Indexed: true
 c      | date    |       | Accumulo column c:c. Indexed: true

2. ZooKeeper CLIを使用して、対応するznodeを削除します。これは、デフォルトのZooKeeperメタデータルート/presto-accumuloを使用することに注意してください。

$ zkCli.sh
[zk: localhost:2181(CONNECTED) 1] delete /presto-accumulo/foo/bar

3. 以前と同じDDLを使用してテーブルを再作成しますが、external=trueプロパティを追加します。以前にcolumn_mappingを定義していなかった場合は、新しいDDLにプロパティを追加する必要があることに注意してください（外部テーブルではこのプロパティを設定する必要があります）。列マッピングはDESCRIBEステートメントの出力にあります。

CREATE TABLE foo.bar (
  a VARCHAR,
  b BIGINT,
  c DATE
)
WITH (
  column_mapping = 'a:a:a,b:b:b,c:c:c',
  index_columns = 'b,c',
  external = true
);