CLI のインストールと設定 - DataHub - Alibaba Cloud ドキュメントセンター

コマンドラインツールを使用して DataHub プロジェクトにアクセスし、コマンドを実行します。

前提条件

Java 8 以降がインストールされていること。

クライアントのインストールと設定

コマンドラインツールパッケージ datahub_console.tar.gz をダウンロードして展開します。
パッケージを展開すると、bin、conf、lib ディレクトリが作成されます。

conf ディレクトリに移動し、datahub.properties ファイルに ak と endpoint の情報を入力します。

datahub.accessid=
datahub.accesskey=
datahub.endpoint=

パラメーターの詳細は次のとおりです。

パラメーター	必須	説明	例
datahub.accessid	はい	Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID。	N/A
datahub.accesskey	はい	指定された AccessKey ID の AccessKey Secret。	N/A
datahub.endpoint	はい	DataHub サービスのエンドポイント。 DataHub プロジェクトのリージョンとネットワークタイプに応じてエンドポイントを設定します。利用可能なエンドポイントの一覧は、「DataHub エンドポイント」をご参照ください。	https://dh-cn-hangzhou.aliyuncs.com

クライアントの実行

次のいずれかの方法でクライアントを起動します。

方法 1：クライアントのインストール先の bin ディレクトリで、datahubcmd.bat (Windows の場合) をダブルクリックしてクライアントを起動します。DataHub=> プロンプトが表示されると、クライアントは正常に起動しています。
方法 2：システムのコマンドラインウィンドウで、クライアントのインストール先の bin ディレクトリに移動します。datahubcmd (Windows の場合) または sh datahubcmd.sh (Linux または macOS の場合) を実行します。DataHub=> プロンプトが表示されると、クライアントは正常に起動しています。

コマンドヘルプ

コマンドのヘルプを取得するには、次のいずれかの方法を使用します。

方法 1：クライアント内からヘルプ情報を表示します。
- すべてのコマンドを表示：
```
help
```
- キーワードで関連コマンドを表示します。たとえば、トピックの一覧を取得するには、次のように入力します。
```
DataHub=>help lt
NAME
        lt - List topic

SYNOPSYS
        lt [-p] string

OPTIONS
        -p  string
                projectName
                [Mandatory]
```
方法 2：システムのコマンドラインウィンドウで、クライアントのインストール先の bin ディレクトリに移動し、次のコマンドを実行してすべてのコマンドのヘルプを表示します。
```
...\bin>datahubcmd help
```

コマンドリファレンス

プロジェクト

プロジェクトの作成
```
cp -p test_project  -c test_comment
```
- -p：プロジェクト名。
- -c：プロジェクトの説明。
プロジェクトの削除
重要
プロジェクトを削除する前に、トピック、サブスクリプション、データ同期タスクなど、プロジェクト内のすべてのリソースを削除する必要があります。そうしないと、操作は失敗します。
```
dp -p test_project
```
- -p：プロジェクト名。
プロジェクトの一覧表示
```
lp
```

トピック

トピックの作成
```
ct -p test_project -t test_topic -m TUPLE -f [(name,string,true)] -s 3 -l 3 -c test_comment
```
- -p：プロジェクト名。
- -t：トピック名。
- -m：トピックのタイプ。有効な値は BLOB または TUPLE です。
- -f：TUPLE タイプのトピックのスキーマ。形式は [(fieldName,fieldType,isNull)] です。複数のフィールドはカンマで区切ります。
- -s：シャード数。
- -l：データライフサイクル (日数)。1～7 の範囲で指定します。
- -c：トピックの説明。
トピックの削除
```
dt -p test_project -t test_topic
```
- -p：プロジェクト名。
- -t：トピック名。
トピック情報の取得
```
gt -p test_project -t test_topic
```
- -p：プロジェクト名。
- -t：トピック名。
トピックスキーマの JSON ファイルへのエクスポート
```
gts -f filepath -p test_project -t test_topic
```
- -f：JSON ファイルを保存するパス。
- -p：プロジェクト名。
- -t：トピック名。
トピックの一覧表示
```
lt -p test_project
```
- -p：プロジェクト名。
JSON ファイルからのトピック作成
```
rtt -s 3 -l 3 -c test_comment -f filepath -p test_project -t test_topic
```
- -s：シャード数。
- -l：データライフサイクル (日数)。1～7 の範囲で指定します。
- -f：ファイルパス。
- -p：プロジェクト名。
- -t：トピック名。
トピックライフサイクルの更新
```
utl -p test_project -t test_topic -l 3 -c test_comment
```
- -p：プロジェクト名。
- -t：トピック名。
- -l：トピックのライフサイクル。
- -c：トピックの説明。
トピックフィールドの追加
```
anf -p test_project -t test_topic -l [(f1,STRING)]
```
- -p：プロジェクト名。
- -t：トピック名。
- -l：追加する新しいフィールド。形式は [(fieldName,fieldType)] です。

コネクタ

ODPS コネクタの作成
```
coc -p test_project -t test_topic -m SYSTEM_TIME -e odpsEndpoint -op odpsProject -ot odpsTable -oa odpsAccessId -ok odpsAccessKey -tr 60 -c (field1,field2) -tf ds hh mm
```
- -p：プロジェクト名。
- -t：トピック名。
- -m：同期タイプ。ODPS への同期でサポートされているタイプは SYSTEM_TIME、USER_DEFINE、EVENT_TIME、META_TIME です。
- -e：ODPS エンドポイント。クラシックネットワークのエンドポイントを使用する必要があります。
- -op：ODPS プロジェクト名。
- -oa：ODPS にアクセスするための AccessKey ID。
- -ok：ODPS にアクセスするための AccessKey Secret。
- -tr：パーティションの時間間隔 (分)。デフォルト値は 60 です。
- -tf：パーティション形式。ds は日単位、ds hh は時間単位、ds hh mm は分単位のパーティションを表します。
ODPS コネクタへのフィールドの追加
```
acf -p test_project -t test_topic -c connectorId -f fieldName
```
- -p：プロジェクト名。
- -t：トピック名。
- -c：コネクタ ID。[データ同期] タブで確認できます。
- -f：新しいフィールドの名前。
MySQL/RDS コネクタの作成
```
cdc -p test_project -t test_topic -h host -po 3306 -ty mysql -d mysql_database -ta msyql_table -u username -pa password -ht IGNORE -n (field1,field2)
```
- -p：プロジェクト名。
- -t：トピック名。
- -h：ホスト。クラシックネットワークのエンドポイントを使用する必要があります。
- -po：ポート。
- -ty：シンクタイプ。有効な値は次のとおりです。
  - SINK_MYSQL は、データを MySQL に同期するためのコネクタを作成します。
  - SINK_ADS は、データを ADS に同期するためのコネクタを作成します。
- -d：データベース名。
- -ta：テーブル名。
- -u：ユーザー名。
- -pa：パスワード。
- -ht：挿入モード。有効な値は次のとおりです。
  - IGNORE
  - OVERWRITE
- -n：同期するフィールド。例：(field1,field2)。
DataHub コネクタの作成
```
cdhc -p test_project -t test_topic -sp sinkProject -st sinkTopic -m AK -i accessid -k accessKey
```
- -p：プロジェクト名。
- -t：トピック名。
- -sp：データのインポート先となるシンクプロジェクト。
- -st：データのインポート先となるシンクトピック。
- -m：認証タイプ。
- AK：AccessKey を使用して認証します。accessId と accessKey が必要です。
- STS：Security Token Service (STS) を使用して認証します。
FC コネクタの作成
```
cfc -p test_project -t test_topic -e endpoint -s service -f function -au AK -i accessId -k accessKey -n (field1,field2)
```
- -p：プロジェクト名。
- -t：トピック名。
- -e：Function Compute (FC) エンドポイント。クラシックネットワークのエンドポイントを使用する必要があります。
- -s：FC サービス名。
- -f：FC 関数名。
- -au：認証方法。
- AK：AccessKey を使用して認証します。accessId と accessKey が必要です。
- STS：Security Token Service (STS) を使用して認証します。
- -n：同期するフィールド。例：(field1,field2)。
Hologres コネクタの作成
```
chc -p test_project -t test_topic -e endpoint -cl (field,field2) -au AK -hp holoProject -ht holoTopic -i accessId -k accessKey -m Delimiter -l 1 -b false -n (field1,field2)
```
- -p：プロジェクト名。
- -t：トピック名。
- -e：エンドポイント。
- -cl：Hologres に同期するフィールド。
- -au：認証方法。現在、Hologres コネクタでは AK 認証のみがサポートされています。
- -m：解析タイプ。
  - Delimiter：lineDelimiter、parseData、columnDelimiter の各プロパティを指定する必要があります。
  - InformaticaJson：parseData プロパティを指定する必要があります。
OTS コネクタの作成
```
cotsc -p test_project -t test_topic -i accessId -k accessKey -it instanceId -m AK -t table -wm PUT -c (field1,field2)
```
- -p：プロジェクト名。
- -t：トピック名。
- -it：Table Store (OTS) インスタンス名。
- -m：認証タイプ。デフォルトは STS です。
- AK: AccessKey を使用して認証します。accessId と accessKey を指定する必要があります。
- STS: STS を使用して認証します。
- -t：OTS テーブル名。
- -wm：書き込みモード。次の 2 つのモードがサポートされています。
  - PUT
  - UPDATE
- -c：同期するフィールド。例：(field1,field2)。
Elasticsearch コネクタの作成
```
cec -p test_project -t test_es -e endpoint:9200 -i f1 -u user -w password -n (f1,f2) -ty (f1,f2) -r (f1,f2) -c (f1,f2) -ti "1970-01-01 00:00:00"
```
- -p：プロジェクト名。
- -t：トピック名。
- -e：Elasticsearch エンドポイント。内部エンドポイントとポートを internal_endpoint:internal_port の形式で指定する必要があります。
- -i：インデックス。インデックスの詳細は、「Elasticsearch にデータを同期するコネクタの作成」をご参照ください。
- -u：Elasticsearch にアクセスするためのユーザー名。
- -w：Elasticsearch にアクセスするためのパスワード。
- -n：ID プロパティ列。
- -ty：タイププロパティ列。
- -r：ルータープロパティ列。
- -c：インポートするフィールド。
- -ti：同期開始時刻。yyyy-mm-dd hh:mm:ss 形式で指定します。
OSS コネクタの作成
```
csc -p test_project -t test_topic -b bucket -e endpoint -pr ossPrefix -tf ossTimeFormat -tr timeRange -c (f1,f2)
```
- -p：プロジェクト名。
- -t: トピック名。
- -b：Object Storage Service (OSS) バケット名。
- -e：OSS エンドポイント名。
- -pr：OSS に同期されるデータのディレクトリプレフィックス。
- -tf：同期時間形式。たとえば、%Y%m%d%H%M は分単位のパーティションを表します。
- -tr：パーティションの時間間隔。
- -c：同期するフィールド。
コネクタの削除
```
dc -p test_project -t test_topic -c connectorId
```
- -p：プロジェクト名。
- -t：トピック名。
- -c：コネクタ ID。[データ同期] タブで確認できます。
コネクタの詳細の取得
```
gc -p test_project -t test_topic -c connectorId
```
- -p：プロジェクト名。
- -t：トピック名。
- -c：コネクタ ID。[データ同期] タブで確認できます。
トピック内のコネクタの一覧表示
```
lc -p test_project -t test_topic
```
- -p：プロジェクト名。
- -t：トピック名。
コネクタの再起動
```
rc -p test_project -t test_topic -c connectorId
```
- -p：プロジェクト名。
- -t：トピック名。
- -c：コネクタ ID。[データ同期] タブで確認できます。
コネクタの AccessKey の更新
```
uca -p test_project -t test_topic -c connectorId  -a accessId -k accessKey
```
- -p：プロジェクト名。
- -t：トピック名。
- -c：コネクタ ID。[データ同期] タブで確認できます。

シャード

シャードのマージ
```
ms -p test_project -t test_topic -s shardId -a adjacentShardId
```
- -p：プロジェクト名。
- -t：トピック名。
- -s：マージするシャードの ID。
- -a：マージする隣接シャードの ID。
シャードの分割
```
ss -p test_project -t test_topic -s shardId
```
- -p：プロジェクト名。
- -t：トピック名。
- -s：分割するシャードの ID。
トピック内のシャードの一覧表示
```
ls -p test_project -t test_topic
```
- -p：プロジェクト名。
- -t：トピック名。
シャード同期ステータスの取得
```
gcs -p test_project -t test_topic -s shardId -c connectorId
```
- -p：プロジェクト名。
- -t：トピック名。
- -s：シャード ID。
- -c：コネクタ ID。[データ同期] タブで確認できます。
シャードの消費オフセットの取得
```
gso -p test_project -t test_topic -s subid -i shardId
```
- -p：プロジェクト名。
- -t：トピック名。
- -s：サブスクリプション ID。
- -i：シャード ID。

サブスクリプション

サブスクリプションの作成
```
css -p test_project -t test_topic -c comment
```
- -p：プロジェクト名。
- -t：トピック名。
- -c：サブスクリプションの説明。
サブスクリプションの削除
```
dsc -p test_project -t test_topic -s subId
```
- -p：プロジェクト名。
- -t：トピック名。
- -s：サブスクリプション ID。
サブスクリプションの一覧表示
```
lss -p test_project -t test_topic
```
- -p：プロジェクト名。
- -t：トピック名。

データのアップロードとダウンロード

データのアップロード
```
uf -f filepath -p test_project -t test_topic -m "," -n 1000
```
- -f：ファイルパス。注意：Windows パスの場合、バックスラッシュをエスケープする必要があります。例：D:\\test\\test.txt。
- -p：プロジェクト名。
- -t：トピック名。
- -m：テキストの区切り文字。カンマ (,) とスペースがサポートされています。
- -n：1 回のアップロードあたりのバッチサイズ。デフォルトは 1000 です。
例：CSV ファイルのアップロード
この例では、次の形式の CSV ファイルを DataHub にアップロードする方法を示します。
```
1. 0,qe614c760fuk8judu01tn5x055rpt1,true,100.1,14321111111
2. 1,znv1py74o8ynn87k66o32ao4x875wi,true,100.1,14321111111
3. 2,7nm0mtpgo1q0ubuljjjx9b000ybltl,true,100.1,14321111111
4. 3,10t0n6pvonnan16279w848ukko5f6l,true,100.1,14321111111
5. 4,0ub584kw88s6dczd0mta7itmta10jo,true,100.1,14321111111
6. 5,1ltfpf0jt7fhvf0oy4lo8m3z62c940,true,100.1,14321111111
7. 6,zpqsfxqy9379lmcehd7q8kftntrozb,true,100.1,14321111111
8. 7,ce1ga9aln346xcj761c3iytshyzuxg,true,100.1,14321111111
9. 8,k5j2id9a0ko90cykl40s6ojq6gruyi,true,100.1,14321111111
10. 9,ns2zcx9bdip5y0aqd1tdicf7bkdmsm,true,100.1,14321111111
11. 10,54rs9cm1xau2fk66pzyz62tf9tsse4,true,100.1,14321111111
```
この CSV ファイルでは、各行が 1 つのレコードで、フィールドはカンマ (,) で区切られています。ファイルはローカルパス /temp/test.csv に保存されているものとします。対応する DataHub トピックのスキーマは次のとおりです。
フィールド名
タイプ
id
BIGINT
name
STRING
gender
BOOLEAN
salary
DOUBLE
my_time
TIMESTAMP
次のコマンドを実行します。
```
uf -f /temp/test.csv -p test_project -t test_topic -m "," -n 1000
```
データのダウンロード
```
down -p test_project -t test_topic -s shardId -d subId -f filePath -ti "1970-01-01 00:00:00" -l 100 -g 0
```
- -f：ダウンロードしたファイルを保存するパス。Windows パスの場合、バックスラッシュをエスケープする必要があります (例：D:\\test\\test.txt)。
- -p：プロジェクト名。
- -t：トピック名。
- -s：シャード ID。
- -d：サブスクリプション ID。
- -ti：データの読み取りを開始するタイムスタンプ。クライアントは、この時刻以降に作成されたレコードを読み取ります。yyyy-mm-dd hh:mm:ss 形式で指定します。
- -l：1 バッチあたりで読み取るレコード数。
- -g：継続的に読み取るかどうかを指定します。
  - 0：レコードの 1 バッチのみを読み取って停止します。
  - 1：継続的に読み取ります。

よくある質問

Windows でスクリプトの起動に失敗する場合：スクリプトパスに括弧が含まれていることが原因である可能性があります。解決するには、スクリプトへの完全なパスに括弧が含まれていないことを確認してください。

フィールド名	タイプ
id	BIGINT
name	STRING
gender	BOOLEAN
salary	DOUBLE
my_time	TIMESTAMP