Snowflake Connector 1.2 の例 - Mule 4

次の例では、ウォーターマークに基づいてテーブルのすべての行を返します。

この例では一部の項目値に変数を使用しています。次のいずれかを実行できます。

コード内で変数を値に置き換える。
プロパティファイルで各変数の値を指定しておき、コネクタ設定からそのファイルを参照する。

プロパティファイルの使用方法についての詳細は、「プロパティプレースホルダーの設定」を参照してください。

この例を試す前に、Anypoint Studio (Studio) にアクセスし、[Mule Palette (Mule パレット)] ビューに Snowflake Connector のエントリが表示されていることを確認します。表示されない場合、「コネクタを Mule プロジェクトに追加する」の手順に従ってください。

テーブルのすべての行を返す

この例では、On Table Row ソースを使用して、最後に処理されたウォーターマークよりもウォーターマーク値が大きい列を含む行を返す方法を示します。

次のスクリーンショットは、この例のアプリケーションフローを示しています。

始める前に

Java 8、11、または 17
Anypoint Studio 7.5 以降
Mule Runtime Engine (Mule) 4.3.0 以降

On Table Row ソースの追加

On Table Row ソースは、一定間隔でテーブルから選択し、取得した行ごとに 1 つのメッセージを生成します。

[Mule Palette (Mule パレット)] ビューから、[Snowflake] を選択して [On Table Row] ソースをキャンバスにドラッグします。
[Search properties (検索プロパティ)] ウィンドウで、[Connector configuration (コネクタ設定)] 項目の横にある [+] をクリックしてグローバル要素を追加します。

次の値を入力します。

項目 Value (値)

項目	Value (値)
Configuration (設定)	設定の名前。
Config Ref (設定参照)	`Snowflake_Config`
Table (テーブル)	`USERS`
Watermark Column (ウォーターマーク列)	`CREATION_DATE`
Id Column (ID 列)	`ID`
Since (以降)	`2021-12-14 11:00:00.000`

Configuration (設定)

設定の名前。

Config Ref (設定参照)

Snowflake_Config

Table (テーブル)

USERS

Watermark Column (ウォーターマーク列)

CREATION_DATE

Id Column (ID 列)

ID

Since (以降)

2021-12-14 11:00:00.000

[Transform Message] コンポーネントを追加する

[Transform Message] コンポーネントでは、テーブルの行を JSON 形式で出力します。

[Mule Palette (Mule パレット)] ビューで [Core (コア)] を選択して [Transform Message] コンポーネントを [On Table Row] の右側にドラッグします。
[Transform Message] コンポーネントをクリックして、出力形式を application/json に設定します。

Logger コンポーネントを追加する

[Logger] コンポーネントでは、クエリ結果がコンソールに記録されます。

[Mule Palette (Mule パレット)] ビューで [Core (コア)] を選択して [Logger] コンポーネントを [Transform Message] の右側にドラッグします。
次の値を入力します。

項目 Value (値)

Display Name (表示名)

Logger

Message (メッセージ)

payload

レベル

INFO (デフォルト)

項目	Value (値)
Display Name (表示名)	`Logger`
Message (メッセージ)	`payload`
レベル	`INFO (デフォルト)`

この例の XML

この例のフローをすばやく Mule アプリケーションに読み込むには、次のコードを Studio XML エディターに貼り付けます。

<?xml version="1.0" encoding="UTF-8"?>

<mule xmlns:db="http://www.mulesoft.org/schema/mule/db" xmlns:ee="http://www.mulesoft.org/schema/mule/ee/core"
	xmlns:snowflake="http://www.mulesoft.org/schema/mule/snowflake"
	xmlns:salesforce="http://www.mulesoft.org/schema/mule/salesforce" xmlns:http="http://www.mulesoft.org/schema/mule/http" xmlns="http://www.mulesoft.org/schema/mule/core" xmlns:doc="http://www.mulesoft.org/schema/mule/documentation" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd
http://www.mulesoft.org/schema/mule/http http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd
http://www.mulesoft.org/schema/mule/snowflake http://www.mulesoft.org/schema/mule/snowflake/current/mule-snowflake.xsd
http://www.mulesoft.org/schema/mule/ee/core http://www.mulesoft.org/schema/mule/ee/core/current/mule-ee.xsd
http://www.mulesoft.org/schema/mule/db http://www.mulesoft.org/schema/mule/db/current/mule-db.xsd">

	<configuration-properties file="mule-app.properties"/>

	<snowflake:snowflake-config name="Snowflake_Config" >
		<snowflake:snowflake-connection accountName="${config.accountName}" warehouse="${config.warehouse}" database="${config.database}" schema="${config.schema}" user="${config.user}" password="${config.password}" role="${config.role}"/>
	</snowflake:snowflake-config>
	<snowflake:snowflake-config name="keyPairWithPasswordConfig" >
        	<snowflake:keypair-connection user="${config.user}"  accountName="${config.accountName}" warehouse="${config.warehouse}" database="${config.database}" schema="${config.schema}" privateKeyFile="${app.home}/rsa_key_encrypted.p8" privateKeyPassword="${config.user.keypair.password}" role="${config.role}"/>
    </snowflake:snowflake-config>
	<flow name="listener-flow" >
		<snowflake:row-listener doc:name="Row Listener" config-ref="Snowflake_Config" table="USERS" watermarkColumn="CREATION_DATE" idColumn="ID" since="2021-12-14 11:00:00.000">
			<scheduling-strategy >
				<fixed-frequency />
			</scheduling-strategy>
		</snowflake:row-listener>
		<ee:transform doc:name="Transform Message" >
			<ee:message >
				<ee:set-payload ><![CDATA[%dw 2.0
output application/json
---
payload]]></ee:set-payload>
			</ee:message>
		</ee:transform>
		<logger level="INFO" doc:name="Logger" message="#[payload]"/>
	</flow>
</mule>xml
コンテンツを展開する

この例の実行手順

コネクタが設定されていることを確認します。
プロジェクトを保存します。
ターミナルにこのソースの新しいメッセージが表示されます。