ストリーミング取り込みとレプリケーションタスクの設定
データ統合で、ストリーミング取り込みとレプリケーションタスクウィザードを使用して、ストリーミング取り込みとレプリケーションタスクを設定できます。ソースとターゲットを定義し、必要に応じてデータを変換するトランスフォーメーションを定義できます。
ウィザードページで、次の設定タスクを完了します。
- 1タスク名、プロジェクトの場所、ランタイム環境など、タスクの基本情報を定義します。
- 2ソースを設定する。
- 3ターゲットを設定する。
- 4必要に応じて、1つまたは複数のトランスフォーメーションを追加します。
- 5必要に応じて、ランタイムオプションを設定する。
タスクウィザードを使用するときは、ターゲットを設定した後にいつでも［保存］をクリックして作業を保存できます。ウィザードが完了したら、［保存］をクリックしてタスクを保存します。
開始する前に、
始める前にに記載されている条件を満たしていることを確認します。
始める前に
ストリーミング取り込みとレプリケーションタスクを作成する前に、次の条件を満たしていることを確認します。
- •組織にストリーミング取り込みとレプリケーションおよびストリーミング取り込みとレプリケーションパッケージのライセンスがあることを確認してください。
- •ストリーミング取り込みとレプリケーションがSecure Agentで実行されている。
- •ソース接続とターゲット接続が存在する。
基本的なタスク情報の定義
ストリーミング取り込みとレプリケーションタスクの定義を開始するには、最初に、タスク名やプロジェクトまたはプロジェクトフォルダの場所、ランタイム環境など、タスクに関するいくつかの基本情報を入力する必要があります。
1以下のいずれかの方法でタスクウィザードを開始します。
- - ホームページで、［取り込み］パネルをクリックし、［ストリーミング取り込みおよびレプリケーションタスク］を選択します。
- - ［参照］ページまたはホームページのナビゲーションバーで、［新規］をクリックして、［新しいアセット］ダイアログボックスを開きます。次に、［データ取り込みおよびレプリケーション］ > ［ストリーミング取り込みおよびレプリケーションタスク］を選択して、［作成］をクリックします。
注: 組織に、アプリケーション、データベース、ファイル、またはストリーミングの各取り込みおよびレプリケーションのカスタムライセンスがない場合、［データ取り込みおよびレプリケーション］カテゴリは引き続き表示されますが、ライセンスのないタスクタイプの取り込みおよびレプリケーションタスクを選択して設定することはできません。
ストリーミング取り込みとレプリケーションタスクウィザードの［定義］ページが表示されます。
2以下のプロパティを設定します。
プロパティ | 説明 |
|---|
名前 | ストリーミング取り込みとレプリケーションタスクの名前。 ストリーミング取り込みとレプリケーションタスクの名前は組織内で一意である必要があります。タスク名には、英数字、スペース、およびアンダースコアを含めることができます。名前の先頭には英字またはアンダースコアを指定する必要があります。 タスク名は大文字と小文字を区別しません。 |
場所 | タスクを保存するプロジェクトフォルダ。 |
ランタイム環境 | Secure Agentを含むランタイム環境。Secure Agentがタスクを実行します。 ストリーミング取り込みおよびレプリケーションタスクの場合、Cloud Hosted Agentはサポートされておらず、［ランタイム環境］リストに表示されません。サーバーレスランタイム環境もサポートされていません。 |
説明 | オプション。タスクの説明。最大長は4,000文字です。 |
3［次へ］をクリックします。
ソースの設定
ソースを設定するには、ストリーミングデータを取り込むソース接続を選択してから、ソースプロパティを設定します。ソースを設定する前に、ソースへの接続がAdministratorサービスで作成されていることを確認してください。
1［ソース］ページで接続を選択します。
ストリーミング取り込みとレプリケーションタスクは次のソースをサポートします。
- - Amazon Kinesis Streams
- - AMQP
- - Apache Kafka
- - Business 360 Events
- - フラットファイル
- - Google PubSub
- - JMS
- - MQTT
- - OPC UA
- - REST V2
接続タイプは、選択した接続名に基づいて自動的に入力されます。
2選択したソースに基づいて、必要な詳細を入力します。
タスクウィザードの［ソース］タブに表示されるオプションは、選択するソースの種類によって異なります。
3詳細プロパティで、必要な情報を入力します。
4［次へ］をクリックします。
［ターゲット］タブが表示されます。
Amazon Kinesis Streamsソースのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ソース］タブで設定できるAmazon Kinesis Streamsソースのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | Amazon KinesisStreamソース接続の名前。 |
接続タイプ | Amazon Kinesis接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
ストリーム | データの読み取り元のKinesis Streamの名前。 |
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ソース］タブでAmazon Kinesis Streamsソースに設定できる詳細プロパティについて説明します。
プロパティ | 説明 |
|---|
GUIDをDynamoDBテーブル名に追加 | Amazon DynamoDBテーブル名にサフィックスとしてGUIDを追加するかどうかを指定します。無効になっている場合は、Amazon DynamoDBテーブル名を入力する必要があります。デフォルトでは有効になっています。 |
Amazon DynamoDB | Kinesisソースデータのチェックポイントの詳細を保存するAmazon DynamoDBテーブル名。 Amazon DynamoDBテーブル名は自動的に生成されます。ただし、選択した名前を入力すると、ストリーミング取り込みとレプリケーションタスクは、自動生成された名前の前に指定された名前をプレフィックスとして付けます。 |
Kinesis Streamsの詳細については、Amazon Web Servicesのドキュメントを参照してください。
AMQPソースのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ソース］タブで設定できるAdvanced Message Queuing Protocol（AMQP）ソースのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | AMQPソース接続の名前。 |
接続タイプ | AMQP接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
キュー | ストリーミング取り込みとレプリケーションタスクがメッセージを読み取る既存のAMQPキューの名前。このキューは、AMQP管理者によって事前定義されています。 |
自動確認メッセージ | TrueまたはFalseを選択できます。Trueを選択すると、AMQPブローカは受信したメッセージを自動的に確認します。 |
バッチサイズ | 1回のセッションでプルする必要のあるメッセージの最大数。 デフォルトは10メッセージです。 |
Azure Event Hubs Kafkaソースのプロパティ
Azure Event Hubs名前空間を使用してKafka接続を作成できます。
標準層または専用層のEvent Hubs名前空間を作成すると、名前空間のKafkaエンドポイントがデフォルトで有効になります。次に、Azure Event Hubs対応のKafka接続を、ストリーミング取り込みとレプリケーションタスクの作成中にソース接続として使用できます。トピック名としてEvent Hubs名を入力します。
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ソース］タブで設定できるKafkaソースのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | Kafkaソース接続の名前。 |
接続タイプ | Kafka接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
トピック | イベントの読み取り元のEvent Hubsの名前。 トピック名を手動で入力するか、Kafka対応のEvent Hubs接続の作成済みメタデータを取得できます。 - 1［選択］をクリックします。
使用可能なすべてのトピックを示す［ソースオブジェクトの選択］ダイアログボックスが表示されます。 - 2必要なトピックを選択して［OK］をクリックします。
|
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ソース］タブでKafkaソースに設定できる詳細プロパティについて説明します。
プロパティ | 説明 |
|---|
コンシューマ設定プロパティ | コンシューマがKafkaに接続するための設定プロパティのカンマ区切りリスト。キーと値のペアとして値を指定します。例: key1=value1, key2=value2 Kafkaコンシューマのgroup.idプロパティは自動生成されます。このプロパティは上書きできます。 |
注: Kafka用のEvent Hubsは、標準層と専用層でのみ使用できます。基本層は、Event HubsでKafkaをサポートしていません。
Business 360 Eventsソースのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ソース］タブで設定できるBusiness 360 Eventsソースのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | Business 360 Events接続を選択し、Business 360データストアからイベントを読み取るようにします。 |
接続タイプ | 接続タイプ、つまりBusiness 360 Eventsです。 注: この属性を変更することはできません。 |
ビジネスオブジェクト | データの読み取り元のパブリッシュイベントアセットを選択します。 |
フラットファイルソースのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ソース］タブで設定できるフラットファイルソースのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | フラットファイルソース接続の名前。 |
接続タイプ | フラットファイル接続タイプ。 接続タイプは、選択した接続に基づいて自動的に表示されます。 |
初期開始位置 | テールするファイル内でデータを読み取る際の開始位置。読み取りを開始する次のいずれかの位置を選択できます。 - - ファイルの先頭。テールするファイルの先頭から読み取ります。すでにロールオーバーされているデータは取り込まないでください。
- - 現在の時刻。テールするファイルの最後に更新された部分から読み取ります。すでにロールオーバーされているデータや、テールするファイル内のすでに書き込まれているデータを取り込まないでください。
|
テーリングモード | ロギングパターンに基づいて、1つまたは複数のファイルをテールします。 次のいずれかのモードを選択します。 - - 単一ファイル。1つのファイルのみをテールします。
- - 複数ファイル。ベースディレクトリに示されているすべてのファイルをテールします。このモードでは、正規表現を入力して、テールするファイルを示すことができます。
|
ファイル | 読み取るファイルの名前の絶対パス。 テールするファイルの名前、またはテールするファイルを見つけるための正規表現。複数ファイルモードのベースディレクトリを入力します。 |
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ソース］タブでフラットファイルソースに設定できる詳細プロパティについて説明します。
接続プロパティ | 説明 |
|---|
ローリングファイル名パターン | ロールオーバーするファイルの名前パターン。 テールするファイルがロールオーバーする場合、ファイル名パターンは、ロールオーバーされたファイルを特定するために使用されます。基盤となるストリーミング取り込みSecure Agentはこのファイルパターンを認識します。Secure Agentが再起動し、ファイルがロールオーバーされると、中断したところから再開します。 アスタリスク（*）と疑問符（?）をワイルドカード文字として使用し、ファイルが同じディレクトリにロールオーバーされることを示すことができます。例: ${filename}.log.*。ここで、アスタリスク（*）は、ファイル名に追加される連続するバージョン番号を表します。 |
Google PubSubソースのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ソース］タブで設定できるGoogle PubSubソースのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | Google PubSubソース接続の名前。 |
接続タイプ | Google PubSub接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
サブスクリプション | メッセージのプル元となるGoogle PubSubサービスのサブスクリプションの名前。 Google PubSub接続は、サブスクリプションのプル配信タイプのみをサポートします。 |
バッチサイズ | Cloudサービスがバッチでバンドルするメッセージの最大数。 デフォルトは1です。 |
JMSソースのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ソース］タブで設定できるJMSソースのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | JMSソース接続の名前。 |
接続タイプ | JMS接続タイプ。接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
宛先タイプ | ソースサービスがJMSメッセージを送信する接続先のタイプ。次のいずれかの接続先タイプを選択できます。 - - キュー。JMSプロバイダは、キューに登録されている単一のコンシューマにメッセージを配信します。
- - ［トピック］。JMSプロバイダは、トピックにサブスクライブしているすべてのアクティブなコンシューマにメッセージを配信します。この接続先タイプを使用すると、複数のコンシューマがメッセージを読み取ることができます。
デフォルトは［キュー］です。 |
共有サブスクリプション | 複数のコンシューマが単一のサブスクリプションにアクセスできるようにします。トピックの接続先タイプに適用されます。 デフォルトはfalseです。 |
継続サブスクリプション | 非アクティブなサブスクライバがメッセージを保持し、その保持したメッセージを、サブスクライバが再接続したときに配信できるようにします。トピックの接続先タイプに適用されます。 デフォルトはfalseです。 |
サブスクリプション名 | サブスクリプションの名前。トピックサブスクリプションが共有可能、継続、またはその両方である場合に、トピックの接続先タイプに適用されます。値が指定されていない場合、取り込みサービスは一意のサブスクリプション名を生成します。 |
JMS接続先 | JMSプロバイダがメッセージを配信するキューまたはトピックの名前。 注: JMS WeblogicサーバーとのJMS接続が作成された場合、キューまたはトピックのJMS接続先は、ピリオドで始まってその後にスラッシュが続きます(./）。以下に例を示します。 。/<JMS Server module name>! <Queue or topic name> ストリーミング取り込みおよびレプリケーションをOracle Weblogic JMSサーバーに接続する方法の詳細については、Informaticaナレッジベースの記事「 000186952」を参照してください。 |
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ソース］タブでJMSソースに設定できる詳細プロパティについて説明します。
プロパティ | 説明 |
|---|
クライアントID | オプション。JMS接続を識別する一意の識別子。 非共有の継続サブスクリプションに値が指定されていない場合、ストリーミング取り込みとレプリケーションタスクは一意のクライアントIDを生成します。 |
Kafkaソースのプロパティ
Kafkaをストリーミング取り込みとレプリケーションタスクのソースとして定義している場合、［ソース］タブでKafkaソースの必須プロパティを設定する必要があります。必要に応じて、コンシューマ設定のプロパティのカンマ区切りリストを指定します。
次の表に、Kafkaソースの必須プロパティを示します。
プロパティ | 説明 |
|---|
接続 | Kafkaソース接続の名前。 |
接続タイプ | Kafka接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
トピック | イベントの読み取り元となるKafkaソーストピック名、またはJavaでサポートされているKafkaソーストピック名パターンの正規表現。 トピック名を手動で入力するか、Kafka接続のメタデータを取得できます。Kafka接続のメタデータを選択するには、次のアクションを実行します。 - 1［選択］をクリックします。
［ソースオブジェクトの選択］ダイアログボックスが表示され、Kafkaブローカで使用可能なすべてのトピックまたはトピックパターンが示されます。 - 2トピックを選択して［OK］をクリックします。
注: 新しいKafkaソーストピックを［稼働中］状態のストリーミング取り込みとレプリケーションジョブに追加するとき、新しいトピックからのデータ損失を回避するために、ジョブをすぐに再デプロイするようにしてください。 |
- コンシューマ設定プロパティ
- ［ソース］タブの［詳細プロパティ］セクションにある［コンシューマ設定プロパティ］で、オプションのコンシューマ設定プロパティのカンマ区切りリストを指定できます。キーと値のペアとして値を指定します。
以下の表に、Kafkaソースで設定できるコンシューマ設定プロパティを示します。
プロパティ | 説明 |
|---|
group.id | Kafkaコンシューマが属するコンシューマグループの名前を指定します。Kafkaコンシューマのコンストラクト時にgroup.idが存在しない場合、タスクはコンシューマグループを自動的に作成します。このプロパティは自動生成されます。このプロパティは上書きできます。デフォルトは、key1=value1, key2=value2です。 |
auto.offset.reset | コミットされた位置がない場合、またはオフセットが範囲外の場合のコンシューマの動作を指定します。 次のタイプの自動オフセットリセットを使用できます。 - - 最も古い。オフセット位置をトピックの先頭にリセットします。
- - 最新。オフセット位置をトピックの最新の位置にリセットします。
- - なし。
Kafkaトピックからデータを読み取るか、トピックパターンを使用するとき、メッセージリカバリ中に前回のチェックポイントのオフセットが削除された場合、次に使用可能なオフセットからメッセージをリカバリするには、次のプロパティを指定します。 auto.offset.reset=earliest それ以外の場合は、ストリーミング取り込みとレプリケーションタスクによって、使用可能な最新のオフセットからデータが読み取られます。 |
message-demarcator | Kafkaソースはメッセージをバッチで受信します。特定のトピックとパーティションのすべてのKafkaメッセージを1つのバッチに含めることができます。このプロパティを使用すると、複数のKafkaメッセージの境界として使用する文字列を指定できます。値を指定しない場合、各Kafkaメッセージは単一のイベントとしてトリガされます。 次の区切り文字を境界として使用できます。 - - 改行。新しい改行で新しいコンテンツを区切ります。新しい改行をメッセージの境界として使用するには、次の値を入力します。
message-demarcator=${literal(' '):unescapeXml()} - - カンマ。カンマで新しいコンテンツを区切ります。カンマをメッセージの境界として使用するには、次の値を入力します。
message-demarcator=${literal(','):unescapeXml()} - - セミコロン。セミコロンで新しいコンテンツを区切ります。セミコロンをメッセージの境界として使用するには、次の値を入力します。
message-demarcator=${literal(';'):unescapeXml()} - - タブ。タブで新しいコンテンツを区切ります。タブをメッセージの境界として使用するには、次の値を入力します。
message-demarcator=${literal('	'):unescapeXml()}
|
max.poll.records | ポーリングの1回の呼び出しで返されるレコードの最大数を指定します。 例: max.poll.records=100000 |
MQTTソースのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ソース］タブで設定できるMQTTソースのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | MQTTソース接続の名前。 |
接続タイプ | MQTT接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
トピック | MQTTトピックの名前。 |
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ソース］タブでMQTTソースに設定できる詳細プロパティについて説明します。
接続プロパティ | 説明 |
|---|
クライアントID | オプション。MQTTソースとMQTTブローカ間の接続を識別する一意の識別子。クライアントIDは、MQTTソースがメッセージの処理中にメッセージを格納するために使用するファイルベースの永続ストアです。 クライアントIDを指定しない場合、ストリーミング取り込みとレプリケーションタスクは、MQTT接続で提供されたクライアントIDを使用します。ただし、MQTT接続でもクライアントIDを指定していない場合は、ストリーミング取り込みとレプリケーションタスクは一意のクライアントIDを生成します。 |
最大キューサイズ | オプション。プロセッサが同時にメモリに保存できるメッセージの最大数。 デフォルト値は1024バイトです。 |
OPC UAソースのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ソース］タブで設定できるOPC UAソースのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | OPC UAソース接続の名前。 |
接続タイプ | OPC UA接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
タグリストの指定形式 | タグのリストを指定する形式。 次のいずれかの形式を選択します。 - - タグのリスト。JSON配列として指定された、OPCクライアントによって読み取られるタグのリスト。
- - タグファイルのパス。JSON配列として指定された、OPCクライアントによって読み取られるタグのリストを含むファイル。
|
タグまたはファイルパス | JSON配列として指定される、タグのリスト、または読み取るタグのリストを含むファイルへのパス。 タグまたはファイルパスのリストは2048文字を超えることはできません。 |
最小パブリッシュ間隔 | サブスクリプション通知メッセージの最小パブリッシュ間隔（ミリ秒単位）。 データの急激な変化を検出するには、このプロパティを低い値に設定します。 デフォルトは1,000ミリ秒です。 |
REST V2ソースのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ソース］タブで設定できるREST V2ソースのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | REST V2ソース接続の名前。 |
接続タイプ | REST V2接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
RESTエンドポイント | 入力Swaggerファイルで指定されたRESTエンドポイントのリスト。 これらのエンドポイントは、選択したREST接続に基づいて表示されます。 |
スキーム | Swagger定義で指定されたスキームのリスト。 選択したスキームを使用してURLを作成します。 |
ポーリング間隔 | 2つの連続するREST呼び出しの間隔。 デフォルトは10秒です。 |
失敗した応答コードに対するアクション | 失敗したREST呼び出しに必要なアクション。 以下のいずれかのオプションを選択することができます。 - - アラートを上げる
- - ダウンストリームにルーティングする。応答をダウンストリームプロセッサにルーティングします。
- - 拒否ディレクトリへのルーティング: [ランタイムオプション]ページで設定された拒否ディレクトリに応答をルーティングします。
|
選択したRESTエンドポイントプロパティで定義された操作IDに基づき、パス、クエリやペイロードなどの動的プロパティがREST V2ソースページの下部に表示されます。
- •ヘッダー。REST呼び出しにヘッダーを追加します。
- •パス。Swagger定義で指定されている複数のパスパラメータで構成されます。パスキーを編集することはできません。パスキーに対応する値のみを入力できます。
- •クエリ。クエリパラメータで構成されます。クエリパラメータはパスパラメータに似ています。
- •ペイロード。
- - サンプルペイロード。PUT、POST、またはPATCH要求の要求本文のスキーマを示す読み取り専用のテキストボックス。例: { "name" : "string", "salary" : "string", "age" : "string" }.
- - 本文。PUT、POST、またはPATCH要求の場合に送信される要求本文。サンプルペイロードからサンプル要求本文をコピーして、必要に応じて値を置き換えることができます。
これらのプロパティはいずれも、Swagger仕様ファイルで必須として定義できます。次に、ストリーミング取り込みとレプリケーションREST V2ソース設定中に同じプロパティが必須と見なされます。RESTエンドポイントをSwagger仕様ファイルで定義しない場合、対応するセクションはストリーミング取り込みとレプリケーションREST V2ページには表示されません。
注: ストリーミング取り込みとレプリケーションタスクを設定するときは、指定するSwagger仕様ファイルの絶対パスがランタイム環境で使用可能であることが必要です。
ターゲットの設定
ターゲットを設定するには、ストリーミングデータを転送するターゲット接続を選択してから、ターゲットプロパティを設定します。ターゲットを設定する前に、ターゲットへの接続がAdministratorサービスで作成されていることを確認してください。
1［ターゲット］ページで接続を選択します。
ストリーミング取り込みとレプリケーションタスクは次のターゲットをサポートします。
- - Amazon Kinesis Data Firehose
- - Amazon Kinesis Streams
- - Amazon S3 V2
- - Apache Kafka
- - Databricks
- - フラットファイル
- - Google BigQuery V2
- - Google PubSub
- - Google Cloud Storage V2
- - JDBC V2
- - Microsoft Azure Data Lake Storage Gen2
- - Microsoft Azure Event Hubs
2選択したターゲットに基づいて、必要な詳細を入力します。
タスクウィザードの［ターゲット］タブに表示されるオプションは、選択するターゲットの種類によって異なります。
3詳細プロパティで、必要な情報を入力します。
4次のいずれかの手順に従います。
- - トランスフォーメーションを追加するには、［次へ］をクリックします。
［トランスフォーメーション］タブが表示されます。
- - タスクを保存するには、［保存］をクリックします。
これで、
ストリーミング取り込みとレプリケーションタスクをデプロイできます。
ストリーミング取り込みとレプリケーションタスクのデプロイの詳細については、「
ストリーミング取り込みとレプリケーションタスクのデプロイ」を参照してください。
Amazon Kinesis Data Firehoseターゲットのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるAmazon Kinesis Data Firehoseターゲットのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | Amazon Kinesis Data Firehoseターゲット接続の名前。 |
接続タイプ | Amazon Kinesis接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
ストリーム名/式 | Kinesisストリーム名またはKinesisストリーム名パターンの正規表現。 正規表現には$expression$形式を使用します。$expression$はデータを評価し、一致するデータをキャプチャグループ1に送信します。 |
Kinesis Data Firehoseの詳細については、Amazon Web Servicesのドキュメントを参照してください。
Amazon Kinesis Streamsターゲットのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるAmazon Kinesis Streamsターゲットのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | Amazon KinesisStreamターゲット接続の名前。 |
接続タイプ | Amazon Kinesis接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
ストリーム名/式 | Kinesisストリーム名またはKinesisストリーム名パターンの正規表現。 正規表現には$expression$形式を使用します。$expression$はデータを評価し、一致するデータをキャプチャグループ1に送信します。 |
Kinesis Streamsの詳細については、Amazon Web Servicesのドキュメントを参照してください。
Amazon S3ターゲットのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるAmazon S3ターゲットのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | Amazon S3ターゲット接続の名前。 |
接続タイプ | Amazon S3 V2接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
オブジェクト名/式 | Amazon S3ファイル名またはAmazon S3ファイル名パターンの正規表現。 正規表現には$expression$形式を使用します。$expression$はデータを評価し、一致するデータをキャプチャグループ1に送信します。 |
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるAmazon S3ターゲット詳細プロパティについて説明します。
プロパティ | 説明 |
|---|
パーティション化間隔 | オプション。ストリーミング取り込みタスクがAmazon S3バケットにパーティションを作成する際に従う時間間隔。このオプションを使用するには、［オブジェクト名/式］フィールドで、オブジェクト名に${Timestamp}式を追加する必要があります。 デフォルトは［なし］です。 詳細については、 Amazon S3ターゲットを参照してください。 |
最小アップロードパートサイズ | オプション。複数の独立したパートセットとしてサイズの大きなファイルをアップロードする場合の最小アップロードパートサイズ（MB単位）。このプロパティを使用して、ファイルのロードをAmazon S3に合わせます。 デフォルト値は5120 MBです。 |
マルチパートアップロードしきい値 | オプション。複数のパートにオブジェクトを同時にアップロードするタイミングを決定するマルチパートダウンロードの最小しきい値。 デフォルト値は5120 MBです。 |
Azure Event Hubsターゲットプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるAzure Event Hubsターゲットのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | Azure Event Hubsターゲット接続の名前。 |
接続タイプ | Azure Event Hub接続のタイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
Event Hub | Azure Event Hubsの名前。 |
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるAzure Event Hubsターゲット詳細プロパティについて説明します。
プロパティ | 説明 |
|---|
共有アクセスポリシー名 | オプション。Event Hub名前空間共有アクセスポリシーの名前。 このポリシーは、この接続に関連付けられたすべてのデータオブジェクトに適用される必要があります。 Event Hubから読み取るには、リスン権限が必要です。Event Hubに書き込むには、ポリシーに送信権限が必要です。 |
共有アクセスポリシーのプライマリキー | オプション。Event Hub名前空間共有アクセスポリシーのプライマリキー。 |
Databricksターゲットのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるDatabricksターゲットのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | Databricksターゲット接続の名前。 |
接続タイプ | Databricks接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
ステージングの場所 | ステージングファイルを保存するための相対ディレクトリパス。 - - DatabricksクラスタがAWSにデプロイされている場合は、Amazon S3ステージングバケットの相対パスを使用します。
- - DatabricksクラスタがAzureにデプロイされている場合は、Azure Data Lake Store Gen2ステージングファイルシステム名の相対パスを使用します。
|
ターゲットテーブル名 | 追加するDatabricksテーブルの名前。 |
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるDatabricksターゲットの詳細プロパティについて説明します。
プロパティ | 説明 |
|---|
ターゲットデータベース名 | AdministratorのDatabricks接続で指定されたデータベース名を上書きします。 |
Databricksターゲットの場合、ソースメッセージはJSON形式とする必要があります。
注: Databricksターゲットを使用したストリーミング取り込みとレプリケーションジョブで、ソーススキーマを変更して追加のデータカラムを含める場合は、ジョブを再デプロイして変更データキャプチャを含めることをお勧めします。
Databricksターゲットを使用したストリーミング取り込みとレプリケーションタスクでフィルタトランスフォーメーションを使用する場合は、取り込んだデータが有効なJSONデータ形式に準拠していることを確認してください。JSONPathフィルタタイプを使用したフィルタトランスフォーメーションは、受信データを検証します。受信データが有効なJSONデータ形式に準拠していない場合、ストリーミング取り込みとレプリケーションタスクはデータを拒否します。次に、拒否されたデータは、設定された拒否ディレクトリに移動します。拒否ディレクトリがまだ設定されていない場合、拒否されたデータは失われます。
Informaticaでは、Databricksターゲットを含むストリーミング取り込みとレプリケーションタスクでCombinerトランスフォーメーションを使用することをお勧めします。ターゲットに書き込む前に、Combinerトランスフォーメーションを追加します。ストリーミング取り込みとレプリケーションタスクは、Databricksターゲットに書き込む前に、ステージングされたすべてのデータを結合します。パフォーマンスを最適化するには、100件以上のレコードをバッチ処理します。
フラットファイルターゲットのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるフラットファイルターゲットのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | フラットファイルターゲット接続の名前。 |
接続タイプ | フラットファイル接続タイプ。 接続タイプは、選択した接続に基づいて表示されます。 |
ステージングディレクトリの場所 | Secure Agent上のステージングディレクトリへのパス。 フラットファイルターゲットにデータを書き込むときに、ファイルをステージングするステージングディレクトリを指定します。ディレクトリに十分なスペースがあり、ディレクトリへの書き込み権限があることを確認します。 |
ロールオーバーサイズ* | タスクがファイルをステージングディレクトリからターゲットに移動する際のファイルサイズ（KB）。 例えば、ロールオーバーサイズを1 MBに設定し、ファイルにtarget.logという名前を付けます。ソースサービスが5 MBをターゲットに送信する場合、ストリーミング取り込みとレプリケーションタスクは最初にtarget.log.<timestamp>ファイルを作成します。target.log.<タイムスタンプ>のサイズが1 MBに達すると、タスクはファイルをロールオーバーします。 |
ロールオーバーイベント数* | ファイルのロールオーバーのために蓄積するイベントまたはメッセージの数。 例えば、ロールオーバーイベント数を1000に設定した場合、ファイルのイベントが1000件蓄積されると、タスクはファイルをロールオーバーします。 |
ロールオーバー時間* | ターゲットファイルがロールオーバーする時間の長さ（ミリ秒単位）。この時間が経過すると、ターゲットファイルはロールオーバーします。 例えば、ロールオーバー時間を1時間に設定した場合、ファイルの時間が1時間に達すると、タスクはファイルをロールオーバーします。 |
ファイル名 | タスクがターゲット上に作成するファイルの名前。 |
*ターゲットファイルのロールオーバーを実行するには、少なくとも1つのロールオーバーオプションの値を指定します。 |
Google BigQuery V2ターゲットのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるGoogle BigQuery V2ターゲットのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | Google BigQuery V2ターゲット接続の名前。 |
接続タイプ | Google BigQuery V2接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
データセット名 | Google BigQueryデータセットの名前。データセットはGoogle Cloud Platformになければなりません。 |
テーブル名 | JSON形式でデータを挿入するGoogle BigQueryテーブルの名前。 |
Google Cloud Storageターゲットのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるGoogle Cloud Storageターゲットのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | Google Cloud Storageターゲット接続の名前。 |
接続タイプ | Google Cloud Storageの接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
再試行回数 | ストリーミング取り込みとレプリケーションタスクが、Google Cloud Storageターゲットへの書き込みを再試行する回数。 デフォルトは6です。 |
バケット | Google Cloud Storageにアップロードするオブジェクトを保存、整理、アクセスするためのコンテナ。 |
キー | Google Cloud Storageターゲット接続の名前。 |
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるGoogle Cloud Storageターゲット詳細プロパティについて説明します。
プロパティ | 説明 |
|---|
プロキシホスト | Secure Agentが使用する送信プロキシサーバーのホスト名。 |
プロキシポート | 送信プロキシサーバーのポート番号。 |
コンテンツタイプ | ファイルのコンテンツタイプ。 application.json、multipart、text、htmlなどの任意のMIMEタイプを指定できます。これらの値では大文字と小文字は区別されません。 デフォルトはtextです。 |
オブジェクトACL | アップロードされたオブジェクトに関連付けられたアクセス制御。 次のいずれかの認証のタイプを選択します。 - - 認証された読み取り。バケットまたはオブジェクトの所有者にFULL_CONTROL権限を付与し、認証されたすべてのGoogleアカウント所有者にREAD権限を付与します。
- - バケット所有者フルコントロール。バケットまたはオブジェクトの所有者にフルコントロールのアクセス許可を付与し、認証されたすべてのGoogleアカウント所有者に読み取り権限を付与します。
- - バケット所有者読み取り専用。オブジェクトの所有者にフルコントロールのアクセス許可を付与し、バケット所有者に読み取り権限を付与します。このタイプはオブジェクトでのみ使用します。
- - 非公開。バケットまたはオブジェクトの所有者に、バケットまたはオブジェクトに対するFULL_CONTROL権限を付与します。
- - プロジェクト非公開。ロールに基づいてプロジェクトチームに権限を与えます。チームの一員である人は誰でもREAD権限を持ち、プロジェクト所有者とプロジェクト編集者はFULL_CONTROL権限を持ちます。これは、新しく作成されたバケットのデフォルトのACLです。
- - 公開読み取り専用。バケット所有者にFULL_CONTROL権限を付与し、匿名のすべてのユーザーにREADおよびWRITE権限を付与します。このACLはバケットのみに適用されます。これをバケットに適用すると、インターネット上の誰でも認証せずにオブジェクトを一覧表示、作成、上書き、削除できます。
|
サーバーサイド暗号化キー | Google Cloud Storageバケットのサーバーサイドの暗号化キー。Google Cloud StorageバケットがSSE-KMSで暗号化されている場合に必要です。 |
コンテンツ破棄タイプ | オブジェクトに添付されるRFC-6266コンテンツ廃棄のタイプ。インラインまたは添付のどちらかを選択します。 |
Google PubSubターゲットのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるGoogle PubSubターゲットのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | Google PubSubターゲット接続の名前。 |
接続タイプ | Google PubSub接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
トピック | ターゲットのGoogle PubSubトピックの名前。 |
バッチサイズ | Cloudサービスがバッチでバンドルするメッセージの最大数。 デフォルトは1です。 |
JDBC V2ターゲットプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるJDBC V2ターゲットのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | JDBC V2ターゲット接続の名前。 |
接続タイプ | JDBC V2接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
テーブル名 | JSON形式でデータを挿入するテーブルの名前。 |
Kafkaターゲットのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるKafkaターゲットのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | Kafkaターゲット接続の名前。 |
接続タイプ | Kafka接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
トピック名/式 | Kafkaトピック名またはJavaがサポートするKafkaトピック名パターンの正規表現。 正規表現には$expression$形式を使用します。$expression$はデータを評価し、一致するデータをキャプチャグループ1に送信します。 トピック名を手動で入力するか、Kafka接続の作成済みメタデータを取得できます。 - 1［選択］をクリックします。
［ターゲットオブジェクトの選択］ダイアログボックスが表示され、Kafkaブローカで使用可能なすべてのトピックが示されます。ただし、Kafkaトピック名のパターンはリストに表示されません。 - 2必要なトピックを選択して［OK］をクリックします。
|
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるKafkaターゲット詳細プロパティについて説明します。
プロパティ | 説明 |
|---|
プロデューサ設定プロパティ | プロデューサの設定プロパティ。 |
メタデータフェッチタイムアウト（ミリ秒） | それ以降にメタデータが取得されなくなる時間。 |
バッチフラッシュサイズ（バイト） | ストリーミング取り込みとレプリケーションタスクがそれ以降なターゲットにデータを書き込むイベントのバッチサイズ。 |
Microsoft Azure Data Lake Storage Gen2ターゲットのプロパティ
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるMicrosoft Azure Data Lake Storage Gen2（ADLS Gen2）ターゲットのプロパティについて説明します。
プロパティ | 説明 |
|---|
接続 | Microsoft Azure Data Lake Storage Gen2ターゲット接続の名前。 |
接続タイプ | ADLS Gen2接続タイプ。 接続タイプは、選択した接続名に基づいて自動的に入力されます。 |
書き込みストラテジ | ADLS Gen2ファイルにデータを書き込む操作タイプ。 ファイルがADLS Gen2ストレージに存在する場合は、既存のファイルの上書き、付加、失敗、またはロールオーバーを選択できます。 デフォルトは［付加］です。 - - 付加。ディレクトリ内の既存のファイルにデータを追加します。
- - 上書き。既存のファイル内の既存のデータを削除し、新しく読み取ったデータを挿入します。
- - 失敗。既存のファイルへのデータの書き込みは失敗します。
- - ロールオーバー。データが書き込まれている現在のファイルを閉じ、設定されたロールオーバー値に基づいて新しいファイルを作成します。
|
中間ディレクトリ | ADLS Gen2のステージングディレクトリへのパス。 ADLS Gen2にデータを書き込むときに、ファイルをステージングするステージングディレクトリを指定します。ディレクトリに十分なスペースがあり、ディレクトリへの書き込み権限があることを確認します。 ［書き込みストラテジ］として［ロールオーバー］を選択した場合に適用されます。 ストリーミング取り込みとレプリケーションジョブでADLS Gen2ターゲットを設定しているときに、ロールオーバープロパティの値を指定しないと、ファイルは中間ディレクトリに残ります。ストリーミング取り込みとレプリケーションジョブを停止またはデプロイ解除すると、中間ディレクトリ内のこれらのファイルは、デフォルトでターゲットの場所に移動されます。 |
ロールオーバーサイズ | ロールオーバーをトリガする際のターゲットファイルサイズ（KB）。 ［書き込みストラテジ］として［ロールオーバー］を選択した場合に適用されます。 |
ロールオーバーイベント数 | ロールオーバーのために蓄積するイベントまたはメッセージの数。 ［書き込みストラテジ］として［ロールオーバー］を選択した場合に適用されます。 |
ロールオーバー時間 | ターゲットファイルがロールオーバーする時間の長さ（ミリ秒単位）。この時間が経過すると、ターゲットファイルはロールオーバーします。 ［書き込みストラテジ］として［ロールオーバー］を選択した場合に適用されます。 |
ファイル名/式 | ファイル名またはファイル名パターンの正規表現。 正規表現には$expression$形式を使用します。$expression$はデータを評価し、一致するデータをキャプチャグループ1に送信します。 |
次の表では、ストリーミング取り込みとレプリケーションタスクを定義するときに［ターゲット］タブで設定できるMicrosoft Azure Data Lake Storage Gen2（ADLS Gen2）ターゲット詳細プロパティについて説明します。
プロパティ | 説明 |
|---|
ファイルシステム名のオーバーライド | 接続で提供されたデフォルトのファイルシステム名をオーバーライドします。このファイルシステム名は、実行時にファイルに書き込むために使用されます。 |
ディレクトリのオーバーライド | デフォルトのディレクトリパスをオーバーライドします。 データの書き込みに使用するADLS Gen2ディレクトリ。 デフォルトはルートディレクトリです。 ターゲットの作成時に指定されたディレクトリパスで、接続の作成時に指定されたパスがオーバーライドされます。 |
圧縮形式 | オプション。ストリーミング取り込みとレプリケーションタスクがターゲットファイルにデータを書き込む前に使用する圧縮形式。 次のいずれかの形式を使用します。 - - なし
- - gzip
- - Bzip2
- - Zlib
- - Deflate
デフォルトは［なし］です。 データレイクストレージから圧縮ファイルを読み取るには、その圧縮ファイルに特定の拡張子が付いている必要があります。圧縮ファイルの読み取りに使用された拡張子が有効でない場合、Secure Agentはそのファイルを処理しません。 |
トランスフォーメーションの設定
ストリーミングデータのデータ形式を指定できます。データ形式に基づいて、トランスフォーメーションを設定できます。
1［トランスフォーメーション］ページで、ストリーミングデータの形式を選択します。
ストリーミング取り込みとレプリケーショントランスフォーメーションは次のデータ形式をサポートします。
2選択したデータ形式に基づいて、サポートされているトランスフォーメーションの1つを選択し、設定します。
3複数のトランスフォーメーションを追加するには、［トランスフォーメーションの追加］をクリックします。
- a［トランスフォーメーション］タブで［トランスフォーメーションの追加］をクリックします。
［新しいトランスフォーメーション］ダイアログボックスが表示されます。
- b選択したトランスフォーメーションのタイプに基づいて、必要なプロパティを入力します。
- c［保存］をクリックします。
保存されたトランスフォーメーションは［トランスフォーメーションの詳細］ウィザードの［トランスフォーメーション］の下に表示されます。
4次のいずれかの手順に従います。
トランスフォーメーションの追加
1［トランスフォーメーション］タブで［+］をクリックしてトランスフォーメーションを追加します。
［新しいトランスフォーメーション］ダイアログボックスが表示されます。
2選択したトランスフォーメーションのタイプに基づいて、必要なプロパティを入力します。
3［保存］をクリックします。
保存されたトランスフォーメーションは［トランスフォーメーションの詳細］ウィザードの［トランスフォーメーション］の下に表示されます。
Combinerトランスフォーメーションのプロパティ
次の表では、Combinerトランスフォーメーションに設定できるプロパティについて説明します。
プロパティ | 説明 |
|---|
トランスフォーメーションのタイプ | ［Combiner］を選択します。 |
トランスフォーメーション名 | Combinerトランスフォーメーションの名前。 |
イベントの最小数 | トランスフォーメーション前に収集するイベントの最小数は、イベントを1つのイベントに結合します。 デフォルトは1です。 |
最大集計サイズ | 結合されたイベントの最大サイズ（メガバイト単位）。 指定されていない場合、このトランスフォーメーションは、イベントを結合する前に、他の2つの条件のいずれかが満たされるのを待ちます。 |
制限時間 | イベントを結合するまでの最大待ち時間。 指定されていない場合、このトランスフォーメーションはイベントを結合する前に他の条件を待機するか、永久に待機します。 |
区切り文字 | 変換されたデータのデータ文字列間の分割を指定するために使用される記号。 バイナリデータ形式にのみ適用されます。 |
各バッチの最終レコードの末尾に区切り文字を追加 | イベントまたはレコードを含むバッチが多数ある場合は、各バッチの最後のレコードの末尾に区切り文字を使用するかどうかを選択できます。これにより、区切り文字が各バッチ間の境界として機能できるようになります。 |
Filterトランスフォーメーションのプロパティ
次の表では、フィルタトランスフォーメーションに設定できるプロパティについて説明します。
プロパティ | 説明 |
|---|
トランスフォーメーションのタイプ | ［フィルタ］を選択します。 |
トランスフォーメーション名 | フィルタトランスフォーメーションの名前。 |
フィルタタイプ | 受信データを評価するフィルタのタイプ。 次のいずれかのフィルタタイプを使用します。 - - JSONパス。JSONプロパティのシーケンスで構成される式。
- - 正規表現。値の範囲またはパターン。
- - XPath。XMLドキュメント内のノードまたはノードセットを選択する式。
|
式 | 選択したフィルタタイプの式。 |
形式変換トランスフォーメーションのプロパティ
ストリーミング取り込みとレプリケーションタスクを定義して、形式変換トランスフォーメーションを追加するときに、タスクウィザードの［新しいトランスフォーメーション］ページでトランスフォーメーションプロパティの値を指定します。
次の表では、形式変換トランスフォーメーションに設定できるプロパティについて説明します。
プロパティ | 説明 |
|---|
トランスフォーメーションのタイプ | ［形式変換］を選択します。 |
トランスフォーメーション名 | 形式変換トランスフォーメーションの名前。 |
形式に変換 | ストリーミング取り込みとレプリケーションタスクは、受信データを選択した形式に変換します。現在、形式変換トランスフォーメーションは、受信データをParquet形式のみに変換します。 |
日付形式* | 入力フィールドに日付の形式を入力します。例: MM/dd/yyyy。 |
時間形式* | 入力フィールドに時間の形式を入力します。例: HH/mm/ss。 |
タイムスタンプの形式* | 入力フィールドにタイムスタンプの形式を入力します。 例えば、10/11/2021 12:04:41 GMT（MM/dd/yyyy HH:mm:ss）のエポックタイムスタンプは1633953881であり、ミリ秒単位のタイムスタンプは1633953881000です。 |
レコードを配列と想定 | 単一のレコードとレコードの配列のどちらを想定するかを決定します。各レコードで配列を想定するには、このプロパティを選択します。XML受信メッセージにのみ適用されます。デフォルトでは、このプロパティの選択は解除されています。 |
*形式が指定されていない場合は、エポック（1970年1月1日午前0時GMT）からのミリ秒単位と見なされます。 |
Javaトランスフォーメーションプロパティ
次の表では、Javaトランスフォーメーションに設定できるプロパティについて説明します。
プロパティ | 説明 |
|---|
トランスフォーメーションのタイプ | Javaを選択します。 |
トランスフォーメーション名 | Javaトランスフォーメーション名。 |
クラスパス | Javaコードの実行に使用されるJARファイル。区切り文字を使用して、複数のJARファイルを含めることができます。UNIXの場合、複数のクラスパスの各項目を区切るにはコロンを使用します。Windowsの場合、複数のクラスパスの各項目を区切るにはセミコロンを使用します。 例: /home/user/commons-text-1.9.jar; /home/user/json-simple-1.1.1.jar |
コードをインポート | サードパーティ、組み込み、およびカスタムのJavaパッケージをインポートします。複数のパッケージをインポートできます。複数のパッケージを区切るにはセミコロン（;）を使用します。次の構文を使用して、パッケージをインポートできます。 import <package name> 例: import java.io.*; |
メインコード | トランスフォーメーションロジックを提供するJavaコード。 例: JSONParser parser = new JSONParser();
try {
JSONObject object = (JSONObject) parser.parse(inputData);
object.put("age", 23);
outputData=object.toJSONString();
} catch (ParseException e) {
throw new RuntimeException();
} |
Joltトランスフォーメーションプロパティ
次の表では、Joltトランスフォーメーションに設定する必要があるプロパティについて説明します。
プロパティ | 説明 |
|---|
トランスフォーメーションのタイプ | Joltを選択します。 |
トランスフォーメーション名 | Joltトランスフォーメーション名。 |
Joltの仕様 | JSON構造を入力し、次の操作を使用して複数の操作のチェーンを追加します。 - - shift。入力JSONの値を読み取り、その値を出力JSONの指定された場所に追加します。
例: [
{
"operation": "shift",
"spec": {
"breadbox": "counterTop"
}
}
] - - default。値または値の配列を出力JSONに追加します。
例: [
{
"operation": "default",
"spec": {
"counterTop": {
"loaf1": {
"slices": [
"slice1",
"slice2",
"slice3",
"slice4
]
}
}
}
}
]
}
- - cardinality。入力JSONの要素を出力JSONの単一の値または配列（リスト）に変換します。
例: [
{
"operation": "cardinality",
"spec": {
"counterTop": {
"loaf1": {
"slices": "ONE"
}
}
}
}
]
- - remove。入力JSONで見つかった場合、要素を削除します。
例: [
{
"operation": "remove",
"spec": {
"counterTop": {
"loaf2": "",
"jar1": ""
}
}
}
]
- - modify。計算された値を出力JSONの要素に書き込みます。
例: [
{
"operation": "modify-overwrite-beta",
"spec": {
"counterTop": {
"jar2": {
"contents": "=toUpper"
}
}
}
}
]
- - sort。すべての配列とマップを入力JSONから出力JSONに並べ替えます。
例: [
{
"operation": "sort"
}
]
以下は、複数の操作を含むJoltの仕様の例です。 Input: {"name":"test"}
[{
"operation": "shift",
"spec": { "name": "testname"
}
}, {
"operation": "default",
"spec": {
"city": ["Anantapur", "Bangalore", "Hyderabad"]
}
}, {
"operation": "cardinality",
"spec": {
"city": "ONE"
}
}, {
"operation": "remove",
"spec": {
"age": ""
}
}, {
"operation": "modify-overwrite-beta",
"spec": {
"city": "=toUpper"
}
}, {
"operation": "sort"}]
注: 入力レコードがJoltの仕様と一致しない場合、トランスフォーメーションはnullレコードをターゲットに書き込みます。 |
Pythonトランスフォーメーションのプロパティ
次の表では、Pythonトランスフォーメーションに設定できるプロパティについて説明します。
プロパティ | 説明 |
|---|
トランスフォーメーションのタイプ | ［Python］を選択します。 |
トランスフォーメーション名 | Pythonトランスフォーメーションの名前。 |
スクリプトの入力タイプ | Pythonスクリプトの入力タイプ。Pythonスクリプトを［スクリプト本文］に入力するか、［スクリプトパス］で利用可能なPythonスクリプトへのパスを指定します。 |
Pythonのパス | Pythonパスライブラリへのディレクトリ。 |
Splitterトランスフォーメーションのプロパティ
次の表では、バイナリメッセージ形式のSplitterトランスフォーメーションに設定できるプロパティについて説明します。
プロパティ | 説明 |
|---|
トランスフォーメーションのタイプ | ［Splitter］を選択します。 |
トランスフォーメーション名 | Splitterトランスフォーメーションの名前。 |
分割タイプ | 条件を分割して、受信データを評価します。 次のいずれかの分割タイプを使用します。 |
行の分割数 | ヘッダー行を除く、各出力分割ファイルに含まれる最大行数。 |
バイトシーケンス | コンテンツを分割するバイトの指定されたシーケンス。 |
次の表では、JSONメッセージ形式のSplitterトランスフォーメーションに設定できるプロパティについて説明します。
プロパティ | 説明 |
|---|
分割式 | 条件を分割して、受信データを評価します。 次のいずれかの分割タイプを使用します。 |
JSONPath式 | JSONまたはスカラーフラグメントに分割する配列要素を指定するJSONPath式。 デフォルトのJSONpath式は$です。 |
次の表では、XMLメッセージ形式のSplitterトランスフォーメーションに設定できるプロパティについて説明します。
プロパティ | 説明 |
|---|
分割の深さ | XMLフラグメントの分割を開始するXMLのネストの深さ。 デフォルトの分割の深さは1です。 |
ランタイムオプションの設定
ストリーミング取り込みとレプリケーションタスクの追加のランタイムオプションを設定できます。ランタイムオプションには、拒否イベントを管理する設定と、エラーについてユーザーに通知する設定が含まれます。
1［ランタイムオプション］ページの［通知管理］で、通知を無効にするか、エラーが発生した場合に通知を受信するまでの時間制限を設定するかを選択します。
時間を分単位または時間単位で設定できます。デフォルトは10分です。
2 タスクがエラーありで実行された場合に通知を送信する電子メールアドレスのリストを入力します。
電子メールアドレスのリストを区切るには、カンマを使用します。組織用に設定された電子メール通知オプションはここでは使用されません。
3［エージェントパラメータ］で、次のタスクを実行します。
- a拒否されたイベントを保存するディレクトリを指定します。
拒否されたイベントはデフォルトでは保存されません。
正規表現でフィルタされたイベントは、拒否ディレクトリには移動されません。
- bログファイルを消去するには、ログファイル消去後のログファイルの最大ファイルサイズを指定します。
ファイルサイズはメガバイトまたはギガバイトで設定できます。デフォルトは10MBです。
- cログに記録するイベントの重要度を選択します。
デフォルトのログレベルは［情報］です。
サポートされているログレベルは次のとおりです。
- ▪ デバッグ。追加のデバッグレベルメッセージとともにすべてのメッセージを記録します。
- ▪ エラー。エラーメッセージのみを記録します。
- ▪ 情報。すべてのエラー、警告、および重要な情報メッセージを記録します。
- ▪ 警告。すべてのエラーと警告メッセージを記録します。
4［詳細パラメータ］で、次のタスクを実行して、ストリーミング取り込みとレプリケーションタスクのパフォーマンスを向上させます。
- aアイコン
（［詳細パラメータ］の横にある）をクリックします。 ［キー］フィールドと［値］フィールドが表示されます。
- b有効なパラメータプロパティとその値を指定します。
5タスクを保存するには、［保存］をクリックします。
これで、
ストリーミング取り込みとレプリケーションタスクをデプロイできます。
ストリーミング取り込みとレプリケーションタスクのデプロイの詳細については、
ストリーミング取り込みとレプリケーションタスクのデプロイを参照してください。
ストリーミング取り込みとレプリケーションタスクのデプロイ
ストリーミング取り込みとレプリケーションタスクを作成した後、タスクをジョブとして実行するには、Secure Agentにデプロイする必要があります。タスクをデプロイする前に、タスク用に設定されたSecure Agentが実行されていることを確認してください。
オブジェクトをデプロイするには、次のいずれかのアクションを実行します。 - - タスクを保存したら、［デプロイ］をクリックします。
- - データ統合の［参照］ページで、タスクを含むプロジェクトを開き、タスクの［アクション］メニューから［デプロイ］を選択します。
タスクが正常にデプロイされたことを示すメッセージが表示されます。これで、ジョブは実行のためのキューに入れられました。
ジョブを編集またはデプロイ解除しても、ジョブを再度デプロイしても、同じSecure AgentまたはSecure Agentグループで実行されている他のジョブには影響しません。
ロールオーバーストラテジを使用するターゲットを含むストリーミング取り込みとレプリケーションジョブを編集または再デプロイすると、定義されたロールオーバー条件を満たしていない場合でも、ステージングディレクトリ内のすべてのファイルがターゲットディレクトリに移動されます。
ストリーミング取り込みとレプリケーションジョブのデプロイ解除
Secure Agentからストリーミング取り込みとレプリケーションジョブをデプロイ解除できます。
データ統合の［参照］ページで、ジョブを含むプロジェクトを開き、ジョブの［アクション］メニューから［デプロイ解除］を選択します。
次のいずれかの監視インタフェースで、デプロイ解除するジョブの行に移動します。
- •ホームページのナビゲーションバーからアクセスする［マイジョブ］ページ
- •モニタの［すべてのジョブ］ページ
- •オペレーションインサイトのデータ取り込みおよびレプリケーションページにある［すべてのジョブ］タブ
ストリーミング取り込みとレプリケーションジョブをデプロイ解除すると、Secure Agentはジョブの以前の状態履歴を保存しません。したがって、ストリーミング取り込みとレプリケーションタスクを再度デプロイすると、タスクは完全に新しいジョブとして実行されます。
ストリーミング取り込みとレプリケーションジョブの停止と再開
ストリーミング取り込みとレプリケーションジョブは停止および再開できます。データ統合の［マイジョブ］ページ、モニタの［すべてのジョブ］ページと［実行中のジョブ］ページ、またはオペレーションインサイトのデータ取り込みおよびレプリケーションページで、ジョブを停止または再開します。
- ジョブを停止します。
稼働中、実行中（エラーあり）、または実行中（警告あり）のストリーミング取り込みとレプリケーションジョブを停止することができます。
ジョブを停止するには、データ統合の［マイジョブ］ページ、またはモニタの［すべてのジョブ］ページと［実行中のジョブ］ページを開き、ジョブの［アクション］メニューから［停止］を選択します。
または、オペレーションインサイトのデータ取り込みおよびレプリケーションページのジョブ行の［アクション］メニューからストリーミング取り込みとレプリケーションジョブを停止することもできます。
［アクション］メニューの横にある［停止］アイコンを使用して、ジョブを停止することもできます。
- ジョブを再開します。
停止されているストリーミング取り込みとレプリケーションジョブを再開できます。
ジョブを再開するには、データ統合の［マイジョブ］ページ、またはモニタの［すべてのジョブ］ページと［実行中のジョブ］ページを開き、ジョブの［アクション］メニューから［再開］を選択します。
または、オペレーションインサイトのデータ取り込みおよびレプリケーションページのジョブ行の［アクション］メニューからストリーミング取り込みとレプリケーションジョブを再開することもできます。
［アクション］メニューの横にある［再開］アイコンを使用して、ジョブを再開することもできます。
ストリーミング取り込みとレプリケーションジョブを停止すると、Secure Agentはジョブの以前の状態履歴を保存します。停止されたストリーミング取り込みとレプリケーションジョブを再開すると、ジョブは最後に保存された状態から実行を開始します。
