Amazon Timestream for LiveAnalytics に類似した機能をご希望の場合は Amazon Timestream for InfluxDB をご検討ください。リアルタイム分析に適した、シンプルなデータインジェストと 1 桁ミリ秒のクエリ応答時間を特徴としています。詳細については、[こちら](https://docs.aws.amazon.com//timestream/latest/developerguide/timestream-for-influxdb.html)を参照してください。
翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
# コードサンプル
AWS SDK を使用して Amazon Timestream にアクセスできます。Timestream は、言語ごとに 2 つの SDK、つまり Write SDK と Query SDK をサポートしています。Write SDK は CRUD オペレーションを実行し、時系列データを Timestream に挿入するために使用されます。Query SDK は、Timestream に保存されている既存の時系列データをクエリするために使用されます。サポートされている各 SDK のコードサンプルなど、詳細については、以下のリストからトピックを選択してください。
**Topics**
+ [Write SDK クライアント](code-samples.write-client.md)
+ [Query SDK クライアント](code-samples.query-client.md)
+ [データベースを作成](code-samples.create-db.md)
+ [データベースを記述する](code-samples.describe-db.md)
+ [データベースを更新する](code-samples.update-db.md)
+ [データベースを削除する](code-samples.delete-db.md)
+ [データベースを一覧表示する](code-samples.list-db.md)
+ [テーブルの作成](code-samples.create-table.md)
+ [テーブルを記述する](code-samples.describe-table.md)
+ [テーブルを更新する](code-samples.update-table.md)
+ [テーブルを削除する](code-samples.delete-table.md)
+ [テーブルの一覧表示](code-samples.list-table.md)
+ [データを書き込む (挿入とアップサート)](code-samples.write.md)
+ [クエリを実行する](code-samples.run-query.md)
+ [UNLOAD クエリを実行する](code-samples.run-query-unload.md)
+ [クエリをキャンセルする](code-samples.cancel-query.md)
+ [バッチロードタスクを作成する](code-samples.create-batch-load.md)
+ [バッチロードタスクを記述する](code-samples.describe-batch-load.md)
+ [バッチロードタスクを一覧表示する](code-samples.list-batch-load-tasks.md)
+ [バッチロードタスクを再開する](code-samples.resume-batch-load-task.md)
+ [スケジュールされたクエリを作成する](code-samples.create-scheduledquery.md)
+ [スケジュールされたクエリを一覧表示する](code-samples.list-scheduledquery.md)
+ [スケジュールされたクエリを記述する](code-samples.describe-scheduledquery.md)
+ [スケジュールされたクエリを実行する](code-samples.execute-scheduledquery.md)
+ [スケジュールされたクエリを更新する](code-samples.update-scheduledquery.md)
+ [スケジュールされたクエリを削除する](code-samples.delete-scheduledquery.md)
# Write SDK クライアント
次のコードスニペットを使用して、Write SDK の Timestream クライアントを作成できます。Write SDK は CRUD オペレーションを実行し、時系列データを Timestream に挿入するために使用されます。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
private static AmazonTimestreamWrite buildWriteClient() {
final ClientConfiguration clientConfiguration = new ClientConfiguration()
.withMaxConnections(5000)
.withRequestTimeout(20 * 1000)
.withMaxErrorRetry(10);
return AmazonTimestreamWriteClientBuilder
.standard()
.withRegion("us-east-1")
.withClientConfiguration(clientConfiguration)
.build();
}
```
------
#### [ Java v2 ]
```
private static TimestreamWriteClient buildWriteClient() {
ApacheHttpClient.Builder httpClientBuilder =
ApacheHttpClient.builder();
httpClientBuilder.maxConnections(5000);
RetryPolicy.Builder retryPolicy =
RetryPolicy.builder();
retryPolicy.numRetries(10);
ClientOverrideConfiguration.Builder overrideConfig =
ClientOverrideConfiguration.builder();
overrideConfig.apiCallAttemptTimeout(Duration.ofSeconds(20));
overrideConfig.retryPolicy(retryPolicy.build());
return TimestreamWriteClient.builder()
.httpClientBuilder(httpClientBuilder)
.overrideConfiguration(overrideConfig.build())
.region(Region.US_EAST_1)
.build();
}
```
------
#### [ Go ]
```
tr := &http.Transport{
ResponseHeaderTimeout: 20 * time.Second,
// Using DefaultTransport values for other parameters: https://golang.org/pkg/net/http/#RoundTripper
Proxy: http.ProxyFromEnvironment,
DialContext: (&net.Dialer{
KeepAlive: 30 * time.Second,
DualStack: true,
Timeout: 30 * time.Second,
}).DialContext,
MaxIdleConns: 100,
IdleConnTimeout: 90 * time.Second,
TLSHandshakeTimeout: 10 * time.Second,
ExpectContinueTimeout: 1 * time.Second,
}
// So client makes HTTP/2 requests
http2.ConfigureTransport(tr)
sess, err := session.NewSession(&aws.Config{ Region: aws.String("us-east-1"), MaxRetries: aws.Int(10), HTTPClient: &http.Client{ Transport: tr }})
writeSvc := timestreamwrite.New(sess)
```
------
#### [ Python ]
```
write_client = session.client('timestream-write', config=Config(read_timeout=20, max_pool_connections = 5000, retries={'max_attempts': 10}))
```
------
#### [ Node.js ]
以下のスニペットでは、AWS SDK for JavaScript v3 を使用しています。クライアントのインストール方法と使用の詳細については、「[Timestream Write クライアント – AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html)」を参照してください。
追加のコマンドのインポートを以下に示します。`CreateDatabaseCommand` インポートはクライアントの作成には必要ありません。
```
import { TimestreamWriteClient, CreateDatabaseCommand } from "@aws-sdk/client-timestream-write";
const writeClient = new TimestreamWriteClient({ region: "us-east-1" });
```
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
var https = require('https');
var agent = new https.Agent({
maxSockets: 5000
});
writeClient = new AWS.TimestreamWrite({
maxRetries: 10,
httpOptions: {
timeout: 20000,
agent: agent
}
});
```
------
#### [ .NET ]
```
var writeClientConfig = new AmazonTimestreamWriteConfig
{
RegionEndpoint = RegionEndpoint.USEast1,
Timeout = TimeSpan.FromSeconds(20),
MaxErrorRetry = 10
};
var writeClient = new AmazonTimestreamWriteClient(writeClientConfig);
```
------
次の設定を使用することをお勧めします。
+ SDK の再試行回数を `10` に設定します。
+ `SDK DEFAULT_BACKOFF_STRATEGY` を使用します。
+ `RequestTimeout` を `20` 秒に設定します。
+ 最大接続数を `5000` 以上に設定します。
# Query SDK クライアント
次のコードスニペットを使用して、Query SDK の Timestream クライアントを作成できます。Query SDK は、Timestream に保存されている既存の時系列データをクエリするために使用されます。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
private static AmazonTimestreamQuery buildQueryClient() {
AmazonTimestreamQuery client = AmazonTimestreamQueryClient.builder().withRegion("us-east-1").build();
return client;
}
```
------
#### [ Java v2 ]
```
private static TimestreamQueryClient buildQueryClient() {
return TimestreamQueryClient.builder()
.region(Region.US_EAST_1)
.build();
}
```
------
#### [ Go ]
```
sess, err := session.NewSession(&aws.Config{Region: aws.String("us-east-1")})
```
------
#### [ Python ]
```
query_client = session.client('timestream-query')
```
------
#### [ Node.js ]
以下のスニペットでは、AWS SDK for JavaScript v3 を使用しています。クライアントのインストール方法と使用状況の詳細については、「[Timestream Query クライアント – ,AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-query/index.html)」を参照してください。
追加のコマンドのインポートを以下に示します。`QueryCommand` インポートはクライアントの作成には必要ありません。
```
import { TimestreamQueryClient, QueryCommand } from "@aws-sdk/client-timestream-query";
const queryClient = new TimestreamQueryClient({ region: "us-east-1" });
```
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
queryClient = new AWS.TimestreamQuery();
```
------
#### [ .NET ]
```
var queryClientConfig = new AmazonTimestreamQueryConfig
{
RegionEndpoint = RegionEndpoint.USEast1
};
var queryClient = new AmazonTimestreamQueryClient(queryClientConfig);
```
------
# データベースを作成
次のコードスニペットを使用してデータベースを作成できます。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
public void createDatabase() {
System.out.println("Creating database");
CreateDatabaseRequest request = new CreateDatabaseRequest();
request.setDatabaseName(DATABASE_NAME);
try {
amazonTimestreamWrite.createDatabase(request);
System.out.println("Database [" + DATABASE_NAME + "] created successfully");
} catch (ConflictException e) {
System.out.println("Database [" + DATABASE_NAME + "] exists. Skipping database creation");
}
}
```
------
#### [ Java v2 ]
```
public void createDatabase() {
System.out.println("Creating database");
CreateDatabaseRequest request = CreateDatabaseRequest.builder().databaseName(DATABASE_NAME).build();
try {
timestreamWriteClient.createDatabase(request);
System.out.println("Database [" + DATABASE_NAME + "] created successfully");
} catch (ConflictException e) {
System.out.println("Database [" + DATABASE_NAME + "] exists. Skipping database creation");
}
}
```
------
#### [ Go ]
```
// Create database.
createDatabaseInput := ×treamwrite.CreateDatabaseInput{
DatabaseName: aws.String(*databaseName),
}
_, err = writeSvc.CreateDatabase(createDatabaseInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Database successfully created")
}
fmt.Println("Describing the database, hit enter to continue")
```
------
#### [ Python ]
```
def create_database(self):
print("Creating Database")
try:
self.client.create_database(DatabaseName=Constant.DATABASE_NAME)
print("Database [%s] created successfully." % Constant.DATABASE_NAME)
except self.client.exceptions.ConflictException:
print("Database [%s] exists. Skipping database creation" % Constant.DATABASE_NAME)
except Exception as err:
print("Create database failed:", err)
```
------
#### [ Node.js ]
以下のスニペットでは、AWS SDK for JavaScript v3 を使用しています。クライアントのインストール方法と使用の詳細については、「[Timestream Write クライアント – AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html)」を参照してください。
「[クラス CreateDatabaseCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/createdatabasecommand.html)」と「[CreateDatabase](https://docs.aws.amazon.com/timestream/latest/developerguide/API_CreateDatabase.html)」も参照してください。
```
import { TimestreamWriteClient, CreateDatabaseCommand } from "@aws-sdk/client-timestream-write";
const writeClient = new TimestreamWriteClient({ region: "us-east-1" });
const params = {
DatabaseName: "testDbFromNode"
};
const command = new CreateDatabaseCommand(params);
try {
const data = await writeClient.send(command);
console.log(`Database ${data.Database.DatabaseName} created successfully`);
} catch (error) {
if (error.code === 'ConflictException') {
console.log(`Database ${params.DatabaseName} already exists. Skipping creation.`);
} else {
console.log("Error creating database", error);
}
}
```
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
async function createDatabase() {
console.log("Creating Database");
const params = {
DatabaseName: constants.DATABASE_NAME
};
const promise = writeClient.createDatabase(params).promise();
await promise.then(
(data) => {
console.log(`Database ${data.Database.DatabaseName} created successfully`);
},
(err) => {
if (err.code === 'ConflictException') {
console.log(`Database ${params.DatabaseName} already exists. Skipping creation.`);
} else {
console.log("Error creating database", err);
}
}
);
}
```
------
#### [ .NET ]
```
public async Task CreateDatabase()
{
Console.WriteLine("Creating Database");
try
{
var createDatabaseRequest = new CreateDatabaseRequest
{
DatabaseName = Constants.DATABASE_NAME
};
CreateDatabaseResponse response = await writeClient.CreateDatabaseAsync(createDatabaseRequest);
Console.WriteLine($"Database {Constants.DATABASE_NAME} created");
}
catch (ConflictException)
{
Console.WriteLine("Database already exists.");
}
catch (Exception e)
{
Console.WriteLine("Create database failed:" + e.ToString());
}
}
```
------
# データベースを記述する
次のコードスニペットを使用して、新しく作成したデータベースの属性に関する情報を取得できます。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
public void describeDatabase() {
System.out.println("Describing database");
final DescribeDatabaseRequest describeDatabaseRequest = new DescribeDatabaseRequest();
describeDatabaseRequest.setDatabaseName(DATABASE_NAME);
try {
DescribeDatabaseResult result = amazonTimestreamWrite.describeDatabase(describeDatabaseRequest);
final Database databaseRecord = result.getDatabase();
final String databaseId = databaseRecord.getArn();
System.out.println("Database " + DATABASE_NAME + " has id " + databaseId);
} catch (final Exception e) {
System.out.println("Database doesn't exist = " + e);
throw e;
}
}
```
------
#### [ Java v2 ]
```
public void describeDatabase() {
System.out.println("Describing database");
final DescribeDatabaseRequest describeDatabaseRequest = DescribeDatabaseRequest.builder()
.databaseName(DATABASE_NAME).build();
try {
DescribeDatabaseResponse response = timestreamWriteClient.describeDatabase(describeDatabaseRequest);
final Database databaseRecord = response.database();
final String databaseId = databaseRecord.arn();
System.out.println("Database " + DATABASE_NAME + " has id " + databaseId);
} catch (final Exception e) {
System.out.println("Database doesn't exist = " + e);
throw e;
}
}
```
------
#### [ Go ]
```
describeDatabaseOutput, err := writeSvc.DescribeDatabase(describeDatabaseInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Describe database is successful, below is the output:")
fmt.Println(describeDatabaseOutput)
}
```
------
#### [ Python ]
```
def describe_database(self):
print("Describing database")
try:
result = self.client.describe_database(DatabaseName=Constant.DATABASE_NAME)
print("Database [%s] has id [%s]" % (Constant.DATABASE_NAME, result['Database']['Arn']))
except self.client.exceptions.ResourceNotFoundException:
print("Database doesn't exist")
except Exception as err:
print("Describe database failed:", err)
```
------
#### [ Node.js ]
以下のスニペットでは、AWS SDK for JavaScript v3 を使用しています。クライアントのインストール方法と使用の詳細については、「[Timestream Write クライアント – AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html)」を参照してください。
「[クラス DescribeDatabaseCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/describedatabasecommand.html)」と「[DescribeDatabase](https://docs.aws.amazon.com/timestream/latest/developerguide/API_DescribeDatabase.html)」も参照してください。
```
import { TimestreamWriteClient, DescribeDatabaseCommand } from "@aws-sdk/client-timestream-write";
const writeClient = new TimestreamWriteClient({ region: "us-east-1" });
const params = {
DatabaseName: "testDbFromNode"
};
const command = new DescribeDatabaseCommand(params);
try {
const data = await writeClient.send(command);
console.log(`Database ${data.Database.DatabaseName} has id ${data.Database.Arn}`);
} catch (error) {
if (error.code === 'ResourceNotFoundException') {
console.log("Database doesn't exist.");
} else {
console.log("Describe database failed.", error);
throw error;
}
}
```
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
async function describeDatabase () {
console.log("Describing Database");
const params = {
DatabaseName: constants.DATABASE_NAME
};
const promise = writeClient.describeDatabase(params).promise();
await promise.then(
(data) => {
console.log(`Database ${data.Database.DatabaseName} has id ${data.Database.Arn}`);
},
(err) => {
if (err.code === 'ResourceNotFoundException') {
console.log("Database doesn't exist.");
} else {
console.log("Describe database failed.", err);
throw err;
}
}
);
}
```
------
#### [ .NET ]
```
public async Task DescribeDatabase()
{
Console.WriteLine("Describing Database");
try
{
var describeDatabaseRequest = new DescribeDatabaseRequest
{
DatabaseName = Constants.DATABASE_NAME
};
DescribeDatabaseResponse response = await writeClient.DescribeDatabaseAsync(describeDatabaseRequest);
Console.WriteLine($"Database {Constants.DATABASE_NAME} has id:{response.Database.Arn}");
}
catch (ResourceNotFoundException)
{
Console.WriteLine("Database does not exist.");
}
catch (Exception e)
{
Console.WriteLine("Describe database failed:" + e.ToString());
}
}
```
------
# データベースを更新する
次のコードスニペットを使用してデータベースを更新できます。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
public void updateDatabase(String kmsId) {
System.out.println("Updating kmsId to " + kmsId);
UpdateDatabaseRequest request = new UpdateDatabaseRequest();
request.setDatabaseName(DATABASE_NAME);
request.setKmsKeyId(kmsId);
try {
UpdateDatabaseResult result = amazonTimestreamWrite.updateDatabase(request);
System.out.println("Update Database complete");
} catch (final ValidationException e) {
System.out.println("Update database failed:");
e.printStackTrace();
} catch (final ResourceNotFoundException e) {
System.out.println("Database " + DATABASE_NAME + " doesn't exist = " + e);
} catch (final Exception e) {
System.out.println("Could not update Database " + DATABASE_NAME + " = " + e);
throw e;
}
}
```
------
#### [ Java v2 ]
```
public void updateDatabase(String kmsKeyId) {
if (kmsKeyId == null) {
System.out.println("Skipping UpdateDatabase because KmsKeyId was not given");
return;
}
System.out.println("Updating database");
UpdateDatabaseRequest request = UpdateDatabaseRequest.builder()
.databaseName(DATABASE_NAME)
.kmsKeyId(kmsKeyId)
.build();
try {
timestreamWriteClient.updateDatabase(request);
System.out.println("Database [" + DATABASE_NAME + "] updated successfully with kmsKeyId " + kmsKeyId);
} catch (ResourceNotFoundException e) {
System.out.println("Database [" + DATABASE_NAME + "] does not exist. Skipping UpdateDatabase");
} catch (Exception e) {
System.out.println("UpdateDatabase failed: " + e);
}
}
```
------
#### [ Go ]
```
// Update Database.
updateDatabaseInput := ×treamwrite.UpdateDatabaseInput {
DatabaseName: aws.String(*databaseName),
KmsKeyId: aws.String(*kmsKeyId),
}
updateDatabaseOutput, err := writeSvc.UpdateDatabase(updateDatabaseInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Update database is successful, below is the output:")
fmt.Println(updateDatabaseOutput)
}
```
------
#### [ Python ]
```
def update_database(self, kms_id):
print("Updating database")
try:
result = self.client.update_database(DatabaseName=Constant.DATABASE_NAME, KmsKeyId=kms_id)
print("Database [%s] was updated to use kms [%s] successfully" % (Constant.DATABASE_NAME,
result['Database']['KmsKeyId']))
except self.client.exceptions.ResourceNotFoundException:
print("Database doesn't exist")
except Exception as err:
print("Update database failed:", err)
```
------
#### [ Node.js ]
以下のスニペットでは、AWS SDK for JavaScript v3 を使用しています。クライアントのインストール方法と使用の詳細については、「[Timestream Write クライアント – AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html)」を参照してください。
「[クラス UpdateDatabaseCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/updatedatabasecommand.html)」および「[UpdateDatabase](https://docs.aws.amazon.com/timestream/latest/developerguide/API_UpdateDatabase.html)」も参照してください。
```
import { TimestreamWriteClient, UpdateDatabaseCommand } from "@aws-sdk/client-timestream-write";
const writeClient = new TimestreamWriteClient({ region: "us-east-1" });
let updatedKmsKeyId = "";
const params = {
DatabaseName: "testDbFromNode",
KmsKeyId: updatedKmsKeyId
};
const command = new UpdateDatabaseCommand(params);
try {
const data = await writeClient.send(command);
console.log(`Database ${data.Database.DatabaseName} updated kmsKeyId to ${updatedKmsKeyId}`);
} catch (error) {
if (error.code === 'ResourceNotFoundException') {
console.log("Database doesn't exist.");
} else {
console.log("Update database failed.", error);
}
}
```
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
async function updateDatabase(updatedKmsKeyId) {
if (updatedKmsKeyId === undefined) {
console.log("Skipping UpdateDatabase; KmsKeyId was not given");
return;
}
console.log("Updating Database");
const params = {
DatabaseName: constants.DATABASE_NAME,
KmsKeyId: updatedKmsKeyId
}
const promise = writeClient.updateDatabase(params).promise();
await promise.then(
(data) => {
console.log(`Database ${data.Database.DatabaseName} updated kmsKeyId to ${updatedKmsKeyId}`);
},
(err) => {
if (err.code === 'ResourceNotFoundException') {
console.log("Database doesn't exist.");
} else {
console.log("Update database failed.", err);
}
}
);
}
```
------
#### [ .NET ]
```
public async Task UpdateDatabase(String updatedKmsKeyId)
{
Console.WriteLine("Updating Database");
try
{
var updateDatabaseRequest = new UpdateDatabaseRequest
{
DatabaseName = Constants.DATABASE_NAME,
KmsKeyId = updatedKmsKeyId
};
UpdateDatabaseResponse response = await writeClient.UpdateDatabaseAsync(updateDatabaseRequest);
Console.WriteLine($"Database {Constants.DATABASE_NAME} updated with KmsKeyId {updatedKmsKeyId}");
}
catch (ResourceNotFoundException)
{
Console.WriteLine("Database does not exist.");
}
catch (Exception e)
{
Console.WriteLine("Update database failed: " + e.ToString());
}
}
private void PrintDatabases(List databases)
{
foreach (Database database in databases)
Console.WriteLine($"Database:{database.DatabaseName}");
}
```
------
# データベースを削除する
次のコードスニペットを使用してデータベースを削除できます。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
public void deleteDatabase() {
System.out.println("Deleting database");
final DeleteDatabaseRequest deleteDatabaseRequest = new DeleteDatabaseRequest();
deleteDatabaseRequest.setDatabaseName(DATABASE_NAME);
try {
DeleteDatabaseResult result =
amazonTimestreamWrite.deleteDatabase(deleteDatabaseRequest);
System.out.println("Delete database status: " + result.getSdkHttpMetadata().getHttpStatusCode());
} catch (final ResourceNotFoundException e) {
System.out.println("Database " + DATABASE_NAME + " doesn't exist = " + e);
throw e;
} catch (final Exception e) {
System.out.println("Could not delete Database " + DATABASE_NAME + " = " + e);
throw e;
}
}
```
------
#### [ Java v2 ]
```
public void deleteDatabase() {
System.out.println("Deleting database");
final DeleteDatabaseRequest deleteDatabaseRequest = new DeleteDatabaseRequest();
deleteDatabaseRequest.setDatabaseName(DATABASE_NAME);
try {
DeleteDatabaseResult result =
amazonTimestreamWrite.deleteDatabase(deleteDatabaseRequest);
System.out.println("Delete database status: " + result.getSdkHttpMetadata().getHttpStatusCode());
} catch (final ResourceNotFoundException e) {
System.out.println("Database " + DATABASE_NAME + " doesn't exist = " + e);
throw e;
} catch (final Exception e) {
System.out.println("Could not delete Database " + DATABASE_NAME + " = " + e);
throw e;
}
}
```
------
#### [ Go ]
```
deleteDatabaseInput := ×treamwrite.DeleteDatabaseInput{
DatabaseName: aws.String(*databaseName),
}
_, err = writeSvc.DeleteDatabase(deleteDatabaseInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Database deleted:", *databaseName)
}
```
------
#### [ Python ]
```
def delete_database(self):
print("Deleting Database")
try:
result = self.client.delete_database(DatabaseName=Constant.DATABASE_NAME)
print("Delete database status [%s]" % result['ResponseMetadata']['HTTPStatusCode'])
except self.client.exceptions.ResourceNotFoundException:
print("database [%s] doesn't exist" % Constant.DATABASE_NAME)
except Exception as err:
print("Delete database failed:", err)
```
------
#### [ Node.js ]
以下のスニペットでは、AWS SDK for JavaScript v3 を使用しています。クライアントのインストール方法と使用の詳細については、「[Timestream Write クライアント – AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html)」を参照してください。
「[クラス DeleteDatabaseCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/deletedatabasecommand.html)」および「[DeleteDatabase](https://docs.aws.amazon.com/timestream/latest/developerguide/API_DeleteDatabase.html)」も参照してください。
```
import { TimestreamWriteClient, DeleteDatabaseCommand } from "@aws-sdk/client-timestream-write";
const writeClient = new TimestreamWriteClient({ region: "us-east-1" });
const params = {
DatabaseName: "testDbFromNode"
};
const command = new DeleteDatabaseCommand(params);
try {
const data = await writeClient.send(command);
console.log("Deleted database");
} catch (error) {
if (error.code === 'ResourceNotFoundException') {
console.log(`Database ${params.DatabaseName} doesn't exists.`);
} else {
console.log("Delete database failed.", error);
throw error;
}
}
```
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
async function deleteDatabase() {
console.log("Deleting Database");
const params = {
DatabaseName: constants.DATABASE_NAME
};
const promise = writeClient.deleteDatabase(params).promise();
await promise.then(
function (data) {
console.log("Deleted database");
},
function(err) {
if (err.code === 'ResourceNotFoundException') {
console.log(`Database ${params.DatabaseName} doesn't exists.`);
} else {
console.log("Delete database failed.", err);
throw err;
}
}
);
}
```
------
#### [ .NET ]
```
public async Task DeleteDatabase()
{
Console.WriteLine("Deleting database");
try
{
var deleteDatabaseRequest = new DeleteDatabaseRequest
{
DatabaseName = Constants.DATABASE_NAME
};
DeleteDatabaseResponse response = await writeClient.DeleteDatabaseAsync(deleteDatabaseRequest);
Console.WriteLine($"Database {Constants.DATABASE_NAME} delete request status:{response.HttpStatusCode}");
}
catch (ResourceNotFoundException)
{
Console.WriteLine($"Database {Constants.DATABASE_NAME} does not exists");
}
catch (Exception e)
{
Console.WriteLine("Exception while deleting database:" + e.ToString());
}
}
```
------
# データベースを一覧表示する
次のコードスニペットを使用してデータベースを一覧表示できます。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
public void listDatabases() {
System.out.println("Listing databases");
ListDatabasesRequest request = new ListDatabasesRequest();
ListDatabasesResult result = amazonTimestreamWrite.listDatabases(request);
final List databases = result.getDatabases();
printDatabases(databases);
String nextToken = result.getNextToken();
while (nextToken != null && !nextToken.isEmpty()) {
request.setNextToken(nextToken);
ListDatabasesResult nextResult = amazonTimestreamWrite.listDatabases(request);
final List nextDatabases = nextResult.getDatabases();
printDatabases(nextDatabases);
nextToken = nextResult.getNextToken();
}
}
private void printDatabases(List databases) {
for (Database db : databases) {
System.out.println(db.getDatabaseName());
}
}
```
------
#### [ Java v2 ]
```
public void listDatabases() {
System.out.println("Listing databases");
ListDatabasesRequest request = ListDatabasesRequest.builder().maxResults(2).build();
ListDatabasesIterable listDatabasesIterable = timestreamWriteClient.listDatabasesPaginator(request);
for(ListDatabasesResponse listDatabasesResponse : listDatabasesIterable) {
final List databases = listDatabasesResponse.databases();
databases.forEach(database -> System.out.println(database.databaseName()));
}
}
```
------
#### [ Go ]
```
// List databases.
listDatabasesMaxResult := int64(15)
listDatabasesInput := ×treamwrite.ListDatabasesInput{
MaxResults: &listDatabasesMaxResult,
}
listDatabasesOutput, err := writeSvc.ListDatabases(listDatabasesInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("List databases is successful, below is the output:")
fmt.Println(listDatabasesOutput)
}
```
------
#### [ Python ]
```
def list_databases(self):
print("Listing databases")
try:
result = self.client.list_databases(MaxResults=5)
self._print_databases(result['Databases'])
next_token = result.get('NextToken', None)
while next_token:
result = self.client.list_databases(NextToken=next_token, MaxResults=5)
self._print_databases(result['Databases'])
next_token = result.get('NextToken', None)
except Exception as err:
print("List databases failed:", err)
```
------
#### [ Node.js ]
以下のスニペットでは、AWS SDK for JavaScript v3 を使用しています。クライアントのインストール方法と使用の詳細については、「[Timestream Write クライアント – AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html)」を参照してください。
「[クラス ListDatabasesCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/listdatabasescommand.html)」と「[ListDatabases](https://docs.aws.amazon.com/timestream/latest/developerguide/API_ListDatabases.html)」も参照してください。
```
import { TimestreamWriteClient, ListDatabasesCommand } from "@aws-sdk/client-timestream-write";
const writeClient = new TimestreamWriteClient({ region: "us-east-1" });
const params = {
MaxResults: 15
};
const command = new ListDatabasesCommand(params);
getDatabasesList(null);
async function getDatabasesList(nextToken) {
if (nextToken) {
params.NextToken = nextToken;
}
try {
const data = await writeClient.send(command);
data.Databases.forEach(function (database) {
console.log(database.DatabaseName);
});
if (data.NextToken) {
return getDatabasesList(data.NextToken);
}
} catch (error) {
console.log("Error while listing databases", error);
}
}
```
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
async function listDatabases() {
console.log("Listing databases:");
const databases = await getDatabasesList(null);
databases.forEach(function(database){
console.log(database.DatabaseName);
});
}
function getDatabasesList(nextToken, databases = []) {
var params = {
MaxResults: 15
};
if(nextToken) {
params.NextToken = nextToken;
}
return writeClient.listDatabases(params).promise()
.then(
(data) => {
databases.push.apply(databases, data.Databases);
if (data.NextToken) {
return getDatabasesList(data.NextToken, databases);
} else {
return databases;
}
},
(err) => {
console.log("Error while listing databases", err);
});
}
```
------
#### [ .NET ]
```
public async Task ListDatabases()
{
Console.WriteLine("Listing Databases");
try
{
var listDatabasesRequest = new ListDatabasesRequest
{
MaxResults = 5
};
ListDatabasesResponse response = await writeClient.ListDatabasesAsync(listDatabasesRequest);
PrintDatabases(response.Databases);
var nextToken = response.NextToken;
while (nextToken != null)
{
listDatabasesRequest.NextToken = nextToken;
response = await writeClient.ListDatabasesAsync(listDatabasesRequest);
PrintDatabases(response.Databases);
nextToken = response.NextToken;
}
}
catch (Exception e)
{
Console.WriteLine("List database failed:" + e.ToString());
}
}
```
------
# テーブルの作成
**Topics**
+ [メモリストアの書き込み](#code-samples.create-table-memorystore)
+ [マグネティックストアの書き込み](#code-samples.create-table-magneticstore)
## メモリストアの書き込み
次のコードスニペットを使用して、マグネティックストアの書き込みが無効になっているテーブルを作成できます。その結果、メモリストアの保持ウィンドウにのみデータを書き込むことができます。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
public void createTable() {
System.out.println("Creating table");
CreateTableRequest createTableRequest = new CreateTableRequest();
createTableRequest.setDatabaseName(DATABASE_NAME);
createTableRequest.setTableName(TABLE_NAME);
final RetentionProperties retentionProperties = new RetentionProperties()
.withMemoryStoreRetentionPeriodInHours(HT_TTL_HOURS)
.withMagneticStoreRetentionPeriodInDays(CT_TTL_DAYS);
createTableRequest.setRetentionProperties(retentionProperties);
try {
amazonTimestreamWrite.createTable(createTableRequest);
System.out.println("Table [" + TABLE_NAME + "] successfully created.");
} catch (ConflictException e) {
System.out.println("Table [" + TABLE_NAME + "] exists on database [" + DATABASE_NAME + "] . Skipping database creation");
}
}
```
------
#### [ Java v2 ]
```
public void createTable() {
System.out.println("Creating table");
final RetentionProperties retentionProperties = RetentionProperties.builder()
.memoryStoreRetentionPeriodInHours(HT_TTL_HOURS)
.magneticStoreRetentionPeriodInDays(CT_TTL_DAYS).build();
final CreateTableRequest createTableRequest = CreateTableRequest.builder()
.databaseName(DATABASE_NAME).tableName(TABLE_NAME).retentionProperties(retentionProperties).build();
try {
timestreamWriteClient.createTable(createTableRequest);
System.out.println("Table [" + TABLE_NAME + "] successfully created.");
} catch (ConflictException e) {
System.out.println("Table [" + TABLE_NAME + "] exists on database [" + DATABASE_NAME + "] . Skipping database creation");
}
}
```
------
#### [ Go ]
```
// Create table.
createTableInput := ×treamwrite.CreateTableInput{
DatabaseName: aws.String(*databaseName),
TableName: aws.String(*tableName),
}
_, err = writeSvc.CreateTable(createTableInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Create table is successful")
}
```
------
#### [ Python ]
```
def create_table(self):
print("Creating table")
retention_properties = {
'MemoryStoreRetentionPeriodInHours': Constant.HT_TTL_HOURS,
'MagneticStoreRetentionPeriodInDays': Constant.CT_TTL_DAYS
}
try:
self.client.create_table(DatabaseName=Constant.DATABASE_NAME, TableName=Constant.TABLE_NAME,
RetentionProperties=retention_properties)
print("Table [%s] successfully created." % Constant.TABLE_NAME)
except self.client.exceptions.ConflictException:
print("Table [%s] exists on database [%s]. Skipping table creation" % (
Constant.TABLE_NAME, Constant.DATABASE_NAME))
except Exception as err:
print("Create table failed:", err)
```
------
#### [ Node.js ]
以下のスニペットでは、AWS SDK for JavaScript v3 を使用しています。クライアントのインストール方法と使用の詳細については、「[Timestream Write クライアント – AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html)」を参照してください。
「[クラス CreateTableCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/createtablecommand.html)」と「[CreateTable](https://docs.aws.amazon.com/timestream/latest/developerguide/API_CreateTable.html)」も参照してください。
```
import { TimestreamWriteClient, CreateTableCommand } from "@aws-sdk/client-timestream-write";
const writeClient = new TimestreamWriteClient({ region: "us-east-1" });
const params = {
DatabaseName: "testDbFromNode",
TableName: "testTableFromNode",
RetentionProperties: {
MemoryStoreRetentionPeriodInHours: 24,
MagneticStoreRetentionPeriodInDays: 365
}
};
const command = new CreateTableCommand(params);
try {
const data = await writeClient.send(command);
console.log(`Table ${data.Table.TableName} created successfully`);
} catch (error) {
if (error.code === 'ConflictException') {
console.log(`Table ${params.TableName} already exists on db ${params.DatabaseName}. Skipping creation.`);
} else {
console.log("Error creating table. ", error);
throw error;
}
}
```
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
async function createTable() {
console.log("Creating Table");
const params = {
DatabaseName: constants.DATABASE_NAME,
TableName: constants.TABLE_NAME,
RetentionProperties: {
MemoryStoreRetentionPeriodInHours: constants.HT_TTL_HOURS,
MagneticStoreRetentionPeriodInDays: constants.CT_TTL_DAYS
}
};
const promise = writeClient.createTable(params).promise();
await promise.then(
(data) => {
console.log(`Table ${data.Table.TableName} created successfully`);
},
(err) => {
if (err.code === 'ConflictException') {
console.log(`Table ${params.TableName} already exists on db ${params.DatabaseName}. Skipping creation.`);
} else {
console.log("Error creating table. ", err);
throw err;
}
}
);
}
```
------
#### [ .NET ]
```
public async Task CreateTable()
{
Console.WriteLine("Creating Table");
try
{
var createTableRequest = new CreateTableRequest
{
DatabaseName = Constants.DATABASE_NAME,
TableName = Constants.TABLE_NAME,
RetentionProperties = new RetentionProperties
{
MagneticStoreRetentionPeriodInDays = Constants.CT_TTL_DAYS,
MemoryStoreRetentionPeriodInHours = Constants.HT_TTL_HOURS
}
};
CreateTableResponse response = await writeClient.CreateTableAsync(createTableRequest);
Console.WriteLine($"Table {Constants.TABLE_NAME} created");
}
catch (ConflictException)
{
Console.WriteLine("Table already exists.");
}
catch (Exception e)
{
Console.WriteLine("Create table failed:" + e.ToString());
}
}
```
------
## マグネティックストアの書き込み
次のコードスニペットを使用して、マグネティックストアの書き込みを有効にしたテーブルを作成できます。マグネティックストア書き込みを使用すると、メモリストア保持ウィンドウとマグネティックストア保持ウィンドウの両方にデータを書き込むことができます。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
public void createTable(String databaseName, String tableName) {
System.out.println("Creating table");
CreateTableRequest createTableRequest = new CreateTableRequest();
createTableRequest.setDatabaseName(databaseName);
createTableRequest.setTableName(tableName);
final RetentionProperties retentionProperties = new RetentionProperties()
.withMemoryStoreRetentionPeriodInHours(HT_TTL_HOURS)
.withMagneticStoreRetentionPeriodInDays(CT_TTL_DAYS);
createTableRequest.setRetentionProperties(retentionProperties);
// Enable MagneticStoreWrite
final MagneticStoreWriteProperties magneticStoreWriteProperties = new MagneticStoreWriteProperties()
.withEnableMagneticStoreWrites(true);
createTableRequest.setMagneticStoreWriteProperties(magneticStoreWriteProperties);
try {
amazonTimestreamWrite.createTable(createTableRequest);
System.out.println("Table [" + tableName + "] successfully created.");
} catch (ConflictException e) {
System.out.println("Table [" + tableName + "] exists on database [" + databaseName + "] . Skipping table creation");
//We do not throw exception here, we use the existing table instead
}
}
```
------
#### [ Java v2 ]
```
public void createTable(String databaseName, String tableName) {
System.out.println("Creating table");
// Enable MagneticStoreWrite
final MagneticStoreWriteProperties magneticStoreWriteProperties =
MagneticStoreWriteProperties.builder()
.enableMagneticStoreWrites(true)
.build();
CreateTableRequest createTableRequest =
CreateTableRequest.builder()
.databaseName(databaseName)
.tableName(tableName)
.retentionProperties(RetentionProperties.builder()
.memoryStoreRetentionPeriodInHours(HT_TTL_HOURS)
.magneticStoreRetentionPeriodInDays(CT_TTL_DAYS)
.build())
.magneticStoreWriteProperties(magneticStoreWriteProperties)
.build();
try {
timestreamWriteClient.createTable(createTableRequest);
System.out.println("Table [" + tableName + "] successfully created.");
} catch (ConflictException e) {
System.out.println("Table [" + tableName + "] exists in database [" + databaseName + "] . Skipping table creation");
}
}
```
------
#### [ Go ]
```
// Create table.
createTableInput := ×treamwrite.CreateTableInput{
DatabaseName: aws.String(*databaseName),
TableName: aws.String(*tableName),
// Enable MagneticStoreWrite
MagneticStoreWriteProperties: ×treamwrite.MagneticStoreWriteProperties{
EnableMagneticStoreWrites: aws.Bool(true),
},
}
_, err = writeSvc.CreateTable(createTableInput)
```
------
#### [ Python ]
```
def create_table(self):
print("Creating table")
retention_properties = {
'MemoryStoreRetentionPeriodInHours': Constant.HT_TTL_HOURS,
'MagneticStoreRetentionPeriodInDays': Constant.CT_TTL_DAYS
}
magnetic_store_write_properties = {
'EnableMagneticStoreWrites': True
}
try:
self.client.create_table(DatabaseName=Constant.DATABASE_NAME, TableName=Constant.TABLE_NAME,
RetentionProperties=retention_properties,
MagneticStoreWriteProperties=magnetic_store_write_properties)
print("Table [%s] successfully created." % Constant.TABLE_NAME)
except self.client.exceptions.ConflictException:
print("Table [%s] exists on database [%s]. Skipping table creation" % (
Constant.TABLE_NAME, Constant.DATABASE_NAME))
except Exception as err:
print("Create table failed:", err)
```
------
#### [ Node.js ]
```
async function createTable() {
console.log("Creating Table");
const params = {
DatabaseName: constants.DATABASE_NAME,
TableName: constants.TABLE_NAME,
RetentionProperties: {
MemoryStoreRetentionPeriodInHours: constants.HT_TTL_HOURS,
MagneticStoreRetentionPeriodInDays: constants.CT_TTL_DAYS
},
MagneticStoreWriteProperties: {
EnableMagneticStoreWrites: true
}
};
const promise = writeClient.createTable(params).promise();
await promise.then(
(data) => {
console.log(`Table ${data.Table.TableName} created successfully`);
},
(err) => {
if (err.code === 'ConflictException') {
console.log(`Table ${params.TableName} already exists on db ${params.DatabaseName}. Skipping creation.`);
} else {
console.log("Error creating table. ", err);
throw err;
}
}
);
}
```
------
#### [ .NET ]
```
public async Task CreateTable()
{
Console.WriteLine("Creating Table");
try
{
var createTableRequest = new CreateTableRequest
{
DatabaseName = Constants.DATABASE_NAME,
TableName = Constants.TABLE_NAME,
RetentionProperties = new RetentionProperties
{
MagneticStoreRetentionPeriodInDays = Constants.CT_TTL_DAYS,
MemoryStoreRetentionPeriodInHours = Constants.HT_TTL_HOURS
},
// Enable MagneticStoreWrite
MagneticStoreWriteProperties = new MagneticStoreWriteProperties
{
EnableMagneticStoreWrites = true,
}
};
CreateTableResponse response = await writeClient.CreateTableAsync(createTableRequest);
Console.WriteLine($"Table {Constants.TABLE_NAME} created");
}
catch (ConflictException)
{
Console.WriteLine("Table already exists.");
}
catch (Exception e)
{
Console.WriteLine("Create table failed:" + e.ToString());
}
}
```
------
# テーブルを記述する
次のコードスニペットを使用して、テーブルの属性に関する情報を取得できます。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
public void describeTable() {
System.out.println("Describing table");
final DescribeTableRequest describeTableRequest = new DescribeTableRequest();
describeTableRequest.setDatabaseName(DATABASE_NAME);
describeTableRequest.setTableName(TABLE_NAME);
try {
DescribeTableResult result = amazonTimestreamWrite.describeTable(describeTableRequest);
String tableId = result.getTable().getArn();
System.out.println("Table " + TABLE_NAME + " has id " + tableId);
} catch (final Exception e) {
System.out.println("Table " + TABLE_NAME + " doesn't exist = " + e);
throw e;
}
}
```
------
#### [ Java v2 ]
```
public void describeTable() {
System.out.println("Describing table");
final DescribeTableRequest describeTableRequest = DescribeTableRequest.builder()
.databaseName(DATABASE_NAME).tableName(TABLE_NAME).build();
try {
DescribeTableResponse response = timestreamWriteClient.describeTable(describeTableRequest);
String tableId = response.table().arn();
System.out.println("Table " + TABLE_NAME + " has id " + tableId);
} catch (final Exception e) {
System.out.println("Table " + TABLE_NAME + " doesn't exist = " + e);
throw e;
}
}
```
------
#### [ Go ]
```
// Describe table.
describeTableInput := ×treamwrite.DescribeTableInput{
DatabaseName: aws.String(*databaseName),
TableName: aws.String(*tableName),
}
describeTableOutput, err := writeSvc.DescribeTable(describeTableInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Describe table is successful, below is the output:")
fmt.Println(describeTableOutput)
}
```
------
#### [ Python ]
```
def describe_table(self):
print("Describing table")
try:
result = self.client.describe_table(DatabaseName=Constant.DATABASE_NAME, TableName=Constant.TABLE_NAME)
print("Table [%s] has id [%s]" % (Constant.TABLE_NAME, result['Table']['Arn']))
except self.client.exceptions.ResourceNotFoundException:
print("Table doesn't exist")
except Exception as err:
print("Describe table failed:", err)
```
------
#### [ Node.js ]
以下のスニペットでは、AWS SDK for JavaScript v3 を使用しています。クライアントのインストール方法と使用の詳細については、「[Timestream Write クライアント – AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html)」を参照してください。
「[クラス DescribeTableCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/describedatabasecommand.html)」と「[DescribeTable](https://docs.aws.amazon.com/timestream/latest/developerguide/API_DescribeTable.html)」も参照してください。
```
import { TimestreamWriteClient, DescribeTableCommand } from "@aws-sdk/client-timestream-write";
const writeClient = new TimestreamWriteClient({ region: "us-east-1" });
const params = {
DatabaseName: "testDbFromNode",
TableName: "testTableFromNode"
};
const command = new DescribeTableCommand(params);
try {
const data = await writeClient.send(command);
console.log(`Table ${data.Table.TableName} has id ${data.Table.Arn}`);
} catch (error) {
if (error.code === 'ResourceNotFoundException') {
console.log("Table or Database doesn't exist.");
} else {
console.log("Describe table failed.", error);
throw error;
}
}
```
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
async function describeTable() {
console.log("Describing Table");
const params = {
DatabaseName: constants.DATABASE_NAME,
TableName: constants.TABLE_NAME
};
const promise = writeClient.describeTable(params).promise();
await promise.then(
(data) => {
console.log(`Table ${data.Table.TableName} has id ${data.Table.Arn}`);
},
(err) => {
if (err.code === 'ResourceNotFoundException') {
console.log("Table or Database doesn't exists.");
} else {
console.log("Describe table failed.", err);
throw err;
}
}
);
}
```
------
#### [ .NET ]
```
public async Task DescribeTable()
{
Console.WriteLine("Describing Table");
try
{
var describeTableRequest = new DescribeTableRequest
{
DatabaseName = Constants.DATABASE_NAME,
TableName = Constants.TABLE_NAME
};
DescribeTableResponse response = await writeClient.DescribeTableAsync(describeTableRequest);
Console.WriteLine($"Table {Constants.TABLE_NAME} has id:{response.Table.Arn}");
}
catch (ResourceNotFoundException)
{
Console.WriteLine("Table does not exist.");
}
catch (Exception e)
{
Console.WriteLine("Describe table failed:" + e.ToString());
}
}
```
------
# テーブルを更新する
次のコードスニペットを使用してテーブルを更新できます。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
public void updateTable() {
System.out.println("Updating table");
UpdateTableRequest updateTableRequest = new UpdateTableRequest();
updateTableRequest.setDatabaseName(DATABASE_NAME);
updateTableRequest.setTableName(TABLE_NAME);
final RetentionProperties retentionProperties = new RetentionProperties()
.withMemoryStoreRetentionPeriodInHours(HT_TTL_HOURS)
.withMagneticStoreRetentionPeriodInDays(CT_TTL_DAYS);
updateTableRequest.setRetentionProperties(retentionProperties);
amazonTimestreamWrite.updateTable(updateTableRequest);
System.out.println("Table updated");
}
```
------
#### [ Java v2 ]
```
public void updateTable() {
System.out.println("Updating table");
final RetentionProperties retentionProperties = RetentionProperties.builder()
.memoryStoreRetentionPeriodInHours(HT_TTL_HOURS)
.magneticStoreRetentionPeriodInDays(CT_TTL_DAYS).build();
final UpdateTableRequest updateTableRequest = UpdateTableRequest.builder()
.databaseName(DATABASE_NAME).tableName(TABLE_NAME).retentionProperties(retentionProperties).build();
timestreamWriteClient.updateTable(updateTableRequest);
System.out.println("Table updated");
}
```
------
#### [ Go ]
```
// Update table.
magneticStoreRetentionPeriodInDays := int64(7 * 365)
memoryStoreRetentionPeriodInHours := int64(24)
updateTableInput := ×treamwrite.UpdateTableInput{
DatabaseName: aws.String(*databaseName),
TableName: aws.String(*tableName),
RetentionProperties: ×treamwrite.RetentionProperties{
MagneticStoreRetentionPeriodInDays: &magneticStoreRetentionPeriodInDays,
MemoryStoreRetentionPeriodInHours: &memoryStoreRetentionPeriodInHours,
},
}
updateTableOutput, err := writeSvc.UpdateTable(updateTableInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Update table is successful, below is the output:")
fmt.Println(updateTableOutput)
}
```
------
#### [ Python ]
```
def update_table(self):
print("Updating table")
retention_properties = {
'MemoryStoreRetentionPeriodInHours': Constant.HT_TTL_HOURS,
'MagneticStoreRetentionPeriodInDays': Constant.CT_TTL_DAYS
}
try:
self.client.update_table(DatabaseName=Constant.DATABASE_NAME, TableName=Constant.TABLE_NAME,
RetentionProperties=retention_properties)
print("Table updated.")
except Exception as err:
print("Update table failed:", err)
```
------
#### [ Node.js ]
以下のスニペットでは、AWS SDK for JavaScript v3 を使用しています。クライアントのインストール方法と使用の詳細については、「[Timestream Write クライアント – AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html)」を参照してください。
また、「[クラス UpdateTableCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/updatetablecommand.html)」と「[UpdateTable](https://docs.aws.amazon.com/timestream/latest/developerguide/API_UpdateTable.html)」も参照してください。
```
import { TimestreamWriteClient, UpdateTableCommand } from "@aws-sdk/client-timestream-write";
const writeClient = new TimestreamWriteClient({ region: "us-east-1" });
const params = {
DatabaseName: "testDbFromNode",
TableName: "testTableFromNode",
RetentionProperties: {
MemoryStoreRetentionPeriodInHours: 24,
MagneticStoreRetentionPeriodInDays: 180
}
};
const command = new UpdateTableCommand(params);
try {
const data = await writeClient.send(command);
console.log("Table updated")
} catch (error) {
console.log("Error updating table. ", error);
}
```
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
async function updateTable() {
console.log("Updating Table");
const params = {
DatabaseName: constants.DATABASE_NAME,
TableName: constants.TABLE_NAME,
RetentionProperties: {
MemoryStoreRetentionPeriodInHours: constants.HT_TTL_HOURS,
MagneticStoreRetentionPeriodInDays: constants.CT_TTL_DAYS
}
};
const promise = writeClient.updateTable(params).promise();
await promise.then(
(data) => {
console.log("Table updated")
},
(err) => {
console.log("Error updating table. ", err);
throw err;
}
);
}
```
------
#### [ .NET ]
```
public async Task UpdateTable()
{
Console.WriteLine("Updating Table");
try
{
var updateTableRequest = new UpdateTableRequest
{
DatabaseName = Constants.DATABASE_NAME,
TableName = Constants.TABLE_NAME,
RetentionProperties = new RetentionProperties
{
MagneticStoreRetentionPeriodInDays = Constants.CT_TTL_DAYS,
MemoryStoreRetentionPeriodInHours = Constants.HT_TTL_HOURS
}
};
UpdateTableResponse response = await writeClient.UpdateTableAsync(updateTableRequest);
Console.WriteLine($"Table {Constants.TABLE_NAME} updated");
}
catch (ResourceNotFoundException)
{
Console.WriteLine("Table does not exist.");
}
catch (Exception e)
{
Console.WriteLine("Update table failed:" + e.ToString());
}
}
```
------
# テーブルを削除する
次のコードスニペットを使用してテーブルを削除できます。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
public void deleteTable() {
System.out.println("Deleting table");
final DeleteTableRequest deleteTableRequest = new DeleteTableRequest();
deleteTableRequest.setDatabaseName(DATABASE_NAME);
deleteTableRequest.setTableName(TABLE_NAME);
try {
DeleteTableResult result =
amazonTimestreamWrite.deleteTable(deleteTableRequest);
System.out.println("Delete table status: " + result.getSdkHttpMetadata().getHttpStatusCode());
} catch (final ResourceNotFoundException e) {
System.out.println("Table " + TABLE_NAME + " doesn't exist = " + e);
throw e;
} catch (final Exception e) {
System.out.println("Could not delete table " + TABLE_NAME + " = " + e);
throw e;
}
}
```
------
#### [ Java v2 ]
```
public void deleteTable() {
System.out.println("Deleting table");
final DeleteTableRequest deleteTableRequest = DeleteTableRequest.builder()
.databaseName(DATABASE_NAME).tableName(TABLE_NAME).build();
try {
DeleteTableResponse response =
timestreamWriteClient.deleteTable(deleteTableRequest);
System.out.println("Delete table status: " + response.sdkHttpResponse().statusCode());
} catch (final ResourceNotFoundException e) {
System.out.println("Table " + TABLE_NAME + " doesn't exist = " + e);
throw e;
} catch (final Exception e) {
System.out.println("Could not delete table " + TABLE_NAME + " = " + e);
throw e;
}
}
```
------
#### [ Go ]
```
deleteTableInput := ×treamwrite.DeleteTableInput{
DatabaseName: aws.String(*databaseName),
TableName: aws.String(*tableName),
}
_, err = writeSvc.DeleteTable(deleteTableInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Table deleted", *tableName)
}
```
------
#### [ Python ]
```
def delete_table(self):
print("Deleting Table")
try:
result = self.client.delete_table(DatabaseName=Constant.DATABASE_NAME, TableName=Constant.TABLE_NAME)
print("Delete table status [%s]" % result['ResponseMetadata']['HTTPStatusCode'])
except self.client.exceptions.ResourceNotFoundException:
print("Table [%s] doesn't exist" % Constant.TABLE_NAME)
except Exception as err:
print("Delete table failed:", err)
```
------
#### [ Node.js ]
以下のスニペットでは、AWS SDK for JavaScript v3 を使用しています。クライアントのインストール方法と使用の詳細については、「[Timestream Write クライアント – AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html)」を参照してください。
「[クラス DeleteTableCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/deletetablecommand.html)」および「[DeleteTable](https://docs.aws.amazon.com/timestream/latest/developerguide/API_DeleteTable.html)」も参照してください。
```
import { TimestreamWriteClient, DeleteTableCommand } from "@aws-sdk/client-timestream-write";
const writeClient = new TimestreamWriteClient({ region: "us-east-1" });
const params = {
DatabaseName: "testDbFromNode",
TableName: "testTableFromNode"
};
const command = new DeleteTableCommand(params);
try {
const data = await writeClient.send(command);
console.log("Deleted table");
} catch (error) {
if (error.code === 'ResourceNotFoundException') {
console.log(`Table ${params.TableName} or Database ${params.DatabaseName} doesn't exist.`);
} else {
console.log("Delete table failed.", error);
throw error;
}
}
```
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
async function deleteTable() {
console.log("Deleting Table");
const params = {
DatabaseName: constants.DATABASE_NAME,
TableName: constants.TABLE_NAME
};
const promise = writeClient.deleteTable(params).promise();
await promise.then(
function (data) {
console.log("Deleted table");
},
function(err) {
if (err.code === 'ResourceNotFoundException') {
console.log(`Table ${params.TableName} or Database ${params.DatabaseName} doesn't exists.`);
} else {
console.log("Delete table failed.", err);
throw err;
}
}
);
}
```
------
#### [ .NET ]
```
public async Task DeleteTable()
{
Console.WriteLine("Deleting table");
try
{
var deleteTableRequest = new DeleteTableRequest
{
DatabaseName = Constants.DATABASE_NAME,
TableName = Constants.TABLE_NAME
};
DeleteTableResponse response = await writeClient.DeleteTableAsync(deleteTableRequest);
Console.WriteLine($"Table {Constants.TABLE_NAME} delete request status: {response.HttpStatusCode}");
}
catch (ResourceNotFoundException)
{
Console.WriteLine($"Table {Constants.TABLE_NAME} does not exists");
}
catch (Exception e)
{
Console.WriteLine("Exception while deleting table:" + e.ToString());
}
}
```
------
# テーブルの一覧表示
次のコードスニペットを使用してテーブルを一覧表示できます。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
public void listTables() {
System.out.println("Listing tables");
ListTablesRequest request = new ListTablesRequest();
request.setDatabaseName(DATABASE_NAME);
ListTablesResult result = amazonTimestreamWrite.listTables(request);
printTables(result.getTables());
String nextToken = result.getNextToken();
while (nextToken != null && !nextToken.isEmpty()) {
request.setNextToken(nextToken);
ListTablesResult nextResult = amazonTimestreamWrite.listTables(request);
printTables(nextResult.getTables());
nextToken = nextResult.getNextToken();
}
}
private void printTables(List tables) {
for (Table table : tables) {
System.out.println(table.getTableName());
}
}
```
------
#### [ Java v2 ]
```
public void listTables() {
System.out.println("Listing tables");
ListTablesRequest request = ListTablesRequest.builder().databaseName(DATABASE_NAME).maxResults(2).build();
ListTablesIterable listTablesIterable = timestreamWriteClient.listTablesPaginator(request);
for(ListTablesResponse listTablesResponse : listTablesIterable) {
final List tables = listTablesResponse.tables();
tables.forEach(table -> System.out.println(table.tableName()));
}
}
```
------
#### [ Go ]
```
listTablesMaxResult := int64(15)
listTablesInput := ×treamwrite.ListTablesInput{
DatabaseName: aws.String(*databaseName),
MaxResults: &listTablesMaxResult,
}
listTablesOutput, err := writeSvc.ListTables(listTablesInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("List tables is successful, below is the output:")
fmt.Println(listTablesOutput)
}
```
------
#### [ Python ]
```
def list_tables(self):
print("Listing tables")
try:
result = self.client.list_tables(DatabaseName=Constant.DATABASE_NAME, MaxResults=5)
self.__print_tables(result['Tables'])
next_token = result.get('NextToken', None)
while next_token:
result = self.client.list_tables(DatabaseName=Constant.DATABASE_NAME,
NextToken=next_token, MaxResults=5)
self.__print_tables(result['Tables'])
next_token = result.get('NextToken', None)
except Exception as err:
print("List tables failed:", err)
```
------
#### [ Node.js ]
以下のスニペットでは、AWS SDK for JavaScript v3 を使用しています。クライアントのインストール方法と使用の詳細については、「[Timestream Write クライアント – AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html)」を参照してください。
「[クラス ListTablesCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/listtablescommand.html)」と「[ListTables](https://docs.aws.amazon.com/timestream/latest/developerguide/API_ListTables.html)」も参照してください。
```
import { TimestreamWriteClient, ListTablesCommand } from "@aws-sdk/client-timestream-write";
const writeClient = new TimestreamWriteClient({ region: "us-east-1" });
const params = {
DatabaseName: "testDbFromNode",
MaxResults: 15
};
const command = new ListTablesCommand(params);
getTablesList(null);
async function getTablesList(nextToken) {
if (nextToken) {
params.NextToken = nextToken;
}
try {
const data = await writeClient.send(command);
data.Tables.forEach(function (table) {
console.log(table.TableName);
});
if (data.NextToken) {
return getTablesList(data.NextToken);
}
} catch (error) {
console.log("Error while listing tables", error);
}
}
```
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
async function listTables() {
console.log("Listing tables:");
const tables = await getTablesList(null);
tables.forEach(function(table){
console.log(table.TableName);
});
}
function getTablesList(nextToken, tables = []) {
var params = {
DatabaseName: constants.DATABASE_NAME,
MaxResults: 15
};
if(nextToken) {
params.NextToken = nextToken;
}
return writeClient.listTables(params).promise()
.then(
(data) => {
tables.push.apply(tables, data.Tables);
if (data.NextToken) {
return getTablesList(data.NextToken, tables);
} else {
return tables;
}
},
(err) => {
console.log("Error while listing databases", err);
});
}
```
------
#### [ .NET ]
```
public async Task ListTables()
{
Console.WriteLine("Listing Tables");
try
{
var listTablesRequest = new ListTablesRequest
{
MaxResults = 5,
DatabaseName = Constants.DATABASE_NAME
};
ListTablesResponse response = await writeClient.ListTablesAsync(listTablesRequest);
PrintTables(response.Tables);
string nextToken = response.NextToken;
while (nextToken != null)
{
listTablesRequest.NextToken = nextToken;
response = await writeClient.ListTablesAsync(listTablesRequest);
PrintTables(response.Tables);
nextToken = response.NextToken;
}
}
catch (Exception e)
{
Console.WriteLine("List table failed:" + e.ToString());
}
}
private void PrintTables(List tables)
{
foreach (Table table in tables)
Console.WriteLine($"Table: {table.TableName}");
}
```
------
# データを書き込む (挿入とアップサート)
**Topics**
+ [レコードのバッチの書き込み](#code-samples.write.write-batches)
+ [共通の属性を持つレコードのバッチの書き込み](#code-samples.write.write-batches-common-attrs)
+ [レコードのアップサート](#code-samples.write.upserts)
+ [マルチメジャー属性の例](#code-samples.write.data.multivalue)
+ [書き込み失敗の処理](#code-samples.write.rejectedRecordException)
## レコードのバッチの書き込み
次のコードスニペットを使用して、Amazon Timestream テーブルにデータを書き込むことができます。バッチでデータを書き込むと、書き込みのコストを最適化できます。詳細については「[書き込み回数の計算](metering-and-pricing.writes.md#metering-and-pricing.writes.write-size-multiple-events)」を参照してください。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
public void writeRecords() {
System.out.println("Writing records");
// Specify repeated values for all records
List records = new ArrayList<>();
final long time = System.currentTimeMillis();
List dimensions = new ArrayList<>();
final Dimension region = new Dimension().withName("region").withValue("us-east-1");
final Dimension az = new Dimension().withName("az").withValue("az1");
final Dimension hostname = new Dimension().withName("hostname").withValue("host1");
dimensions.add(region);
dimensions.add(az);
dimensions.add(hostname);
Record cpuUtilization = new Record()
.withDimensions(dimensions)
.withMeasureName("cpu_utilization")
.withMeasureValue("13.5")
.withMeasureValueType(MeasureValueType.DOUBLE)
.withTime(String.valueOf(time));
Record memoryUtilization = new Record()
.withDimensions(dimensions)
.withMeasureName("memory_utilization")
.withMeasureValue("40")
.withMeasureValueType(MeasureValueType.DOUBLE)
.withTime(String.valueOf(time));
records.add(cpuUtilization);
records.add(memoryUtilization);
WriteRecordsRequest writeRecordsRequest = new WriteRecordsRequest()
.withDatabaseName(DATABASE_NAME)
.withTableName(TABLE_NAME)
.withRecords(records);
try {
WriteRecordsResult writeRecordsResult = amazonTimestreamWrite.writeRecords(writeRecordsRequest);
System.out.println("WriteRecords Status: " + writeRecordsResult.getSdkHttpMetadata().getHttpStatusCode());
} catch (RejectedRecordsException e) {
System.out.println("RejectedRecords: " + e);
for (RejectedRecord rejectedRecord : e.getRejectedRecords()) {
System.out.println("Rejected Index " + rejectedRecord.getRecordIndex() + ": "
+ rejectedRecord.getReason());
}
System.out.println("Other records were written successfully. ");
} catch (Exception e) {
System.out.println("Error: " + e);
}
}
```
------
#### [ Java v2 ]
```
public void writeRecords() {
System.out.println("Writing records");
// Specify repeated values for all records
List records = new ArrayList<>();
final long time = System.currentTimeMillis();
List dimensions = new ArrayList<>();
final Dimension region = Dimension.builder().name("region").value("us-east-1").build();
final Dimension az = Dimension.builder().name("az").value("az1").build();
final Dimension hostname = Dimension.builder().name("hostname").value("host1").build();
dimensions.add(region);
dimensions.add(az);
dimensions.add(hostname);
Record cpuUtilization = Record.builder()
.dimensions(dimensions)
.measureValueType(MeasureValueType.DOUBLE)
.measureName("cpu_utilization")
.measureValue("13.5")
.time(String.valueOf(time)).build();
Record memoryUtilization = Record.builder()
.dimensions(dimensions)
.measureValueType(MeasureValueType.DOUBLE)
.measureName("memory_utilization")
.measureValue("40")
.time(String.valueOf(time)).build();
records.add(cpuUtilization);
records.add(memoryUtilization);
WriteRecordsRequest writeRecordsRequest = WriteRecordsRequest.builder()
.databaseName(DATABASE_NAME).tableName(TABLE_NAME).records(records).build();
try {
WriteRecordsResponse writeRecordsResponse = timestreamWriteClient.writeRecords(writeRecordsRequest);
System.out.println("WriteRecords Status: " + writeRecordsResponse.sdkHttpResponse().statusCode());
} catch (RejectedRecordsException e) {
System.out.println("RejectedRecords: " + e);
for (RejectedRecord rejectedRecord : e.rejectedRecords()) {
System.out.println("Rejected Index " + rejectedRecord.recordIndex() + ": "
+ rejectedRecord.reason());
}
System.out.println("Other records were written successfully. ");
} catch (Exception e) {
System.out.println("Error: " + e);
}
}
```
------
#### [ Go ]
```
now := time.Now()
currentTimeInSeconds := now.Unix()
writeRecordsInput := ×treamwrite.WriteRecordsInput{
DatabaseName: aws.String(*databaseName),
TableName: aws.String(*tableName),
Records: []*timestreamwrite.Record{
×treamwrite.Record{
Dimensions: []*timestreamwrite.Dimension{
×treamwrite.Dimension{
Name: aws.String("region"),
Value: aws.String("us-east-1"),
},
×treamwrite.Dimension{
Name: aws.String("az"),
Value: aws.String("az1"),
},
×treamwrite.Dimension{
Name: aws.String("hostname"),
Value: aws.String("host1"),
},
},
MeasureName: aws.String("cpu_utilization"),
MeasureValue: aws.String("13.5"),
MeasureValueType: aws.String("DOUBLE"),
Time: aws.String(strconv.FormatInt(currentTimeInSeconds, 10)),
TimeUnit: aws.String("SECONDS"),
},
×treamwrite.Record{
Dimensions: []*timestreamwrite.Dimension{
×treamwrite.Dimension{
Name: aws.String("region"),
Value: aws.String("us-east-1"),
},
×treamwrite.Dimension{
Name: aws.String("az"),
Value: aws.String("az1"),
},
×treamwrite.Dimension{
Name: aws.String("hostname"),
Value: aws.String("host1"),
},
},
MeasureName: aws.String("memory_utilization"),
MeasureValue: aws.String("40"),
MeasureValueType: aws.String("DOUBLE"),
Time: aws.String(strconv.FormatInt(currentTimeInSeconds, 10)),
TimeUnit: aws.String("SECONDS"),
},
},
}
_, err = writeSvc.WriteRecords(writeRecordsInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Write records is successful")
}
```
------
#### [ Python ]
```
def write_records(self):
print("Writing records")
current_time = self._current_milli_time()
dimensions = [
{'Name': 'region', 'Value': 'us-east-1'},
{'Name': 'az', 'Value': 'az1'},
{'Name': 'hostname', 'Value': 'host1'}
]
cpu_utilization = {
'Dimensions': dimensions,
'MeasureName': 'cpu_utilization',
'MeasureValue': '13.5',
'MeasureValueType': 'DOUBLE',
'Time': current_time
}
memory_utilization = {
'Dimensions': dimensions,
'MeasureName': 'memory_utilization',
'MeasureValue': '40',
'MeasureValueType': 'DOUBLE',
'Time': current_time
}
records = [cpu_utilization, memory_utilization]
try:
result = self.client.write_records(DatabaseName=Constant.DATABASE_NAME, TableName=Constant.TABLE_NAME,
Records=records, CommonAttributes={})
print("WriteRecords Status: [%s]" % result['ResponseMetadata']['HTTPStatusCode'])
except self.client.exceptions.RejectedRecordsException as err:
self._print_rejected_records_exceptions(err)
except Exception as err:
print("Error:", err)
@staticmethod
def _print_rejected_records_exceptions(err):
print("RejectedRecords: ", err)
for rr in err.response["RejectedRecords"]:
print("Rejected Index " + str(rr["RecordIndex"]) + ": " + rr["Reason"])
if "ExistingVersion" in rr:
print("Rejected record existing version: ", rr["ExistingVersion"])
@staticmethod
def _current_milli_time():
return str(int(round(time.time() * 1000)))
```
------
#### [ Node.js ]
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
async function writeRecords() {
console.log("Writing records");
const currentTime = Date.now().toString(); // Unix time in milliseconds
const dimensions = [
{'Name': 'region', 'Value': 'us-east-1'},
{'Name': 'az', 'Value': 'az1'},
{'Name': 'hostname', 'Value': 'host1'}
];
const cpuUtilization = {
'Dimensions': dimensions,
'MeasureName': 'cpu_utilization',
'MeasureValue': '13.5',
'MeasureValueType': 'DOUBLE',
'Time': currentTime.toString()
};
const memoryUtilization = {
'Dimensions': dimensions,
'MeasureName': 'memory_utilization',
'MeasureValue': '40',
'MeasureValueType': 'DOUBLE',
'Time': currentTime.toString()
};
const records = [cpuUtilization, memoryUtilization];
const params = {
DatabaseName: constants.DATABASE_NAME,
TableName: constants.TABLE_NAME,
Records: records
};
const request = writeClient.writeRecords(params);
await request.promise().then(
(data) => {
console.log("Write records successful");
},
(err) => {
console.log("Error writing records:", err);
if (err.code === 'RejectedRecordsException') {
const responsePayload = JSON.parse(request.response.httpResponse.body.toString());
console.log("RejectedRecords: ", responsePayload.RejectedRecords);
console.log("Other records were written successfully. ");
}
}
);
}
```
------
#### [ .NET ]
```
public async Task WriteRecords()
{
Console.WriteLine("Writing records");
DateTimeOffset now = DateTimeOffset.UtcNow;
string currentTimeString = (now.ToUnixTimeMilliseconds()).ToString();
List dimensions = new List{
new Dimension { Name = "region", Value = "us-east-1" },
new Dimension { Name = "az", Value = "az1" },
new Dimension { Name = "hostname", Value = "host1" }
};
var cpuUtilization = new Record
{
Dimensions = dimensions,
MeasureName = "cpu_utilization",
MeasureValue = "13.6",
MeasureValueType = MeasureValueType.DOUBLE,
Time = currentTimeString
};
var memoryUtilization = new Record
{
Dimensions = dimensions,
MeasureName = "memory_utilization",
MeasureValue = "40",
MeasureValueType = MeasureValueType.DOUBLE,
Time = currentTimeString
};
List records = new List {
cpuUtilization,
memoryUtilization
};
try
{
var writeRecordsRequest = new WriteRecordsRequest
{
DatabaseName = Constants.DATABASE_NAME,
TableName = Constants.TABLE_NAME,
Records = records
};
WriteRecordsResponse response = await writeClient.WriteRecordsAsync(writeRecordsRequest);
Console.WriteLine($"Write records status code: {response.HttpStatusCode.ToString()}");
}
catch (RejectedRecordsException e) {
Console.WriteLine("RejectedRecordsException:" + e.ToString());
foreach (RejectedRecord rr in e.RejectedRecords) {
Console.WriteLine("RecordIndex " + rr.RecordIndex + " : " + rr.Reason);
}
Console.WriteLine("Other records were written successfully. ");
}
catch (Exception e)
{
Console.WriteLine("Write records failure:" + e.ToString());
}
}
```
------
## 共通の属性を持つレコードのバッチの書き込み
時系列データに多くのデータポイントで共通するメジャーやディメンションがある場合は、以下の最適化されたバージョンの writeRecords API を使用し、データを Timestream for LiveAnalytics に挿入することもできます。バッチ処理で一般的な属性を使用すると、「[書き込み回数の計算](metering-and-pricing.writes.md#metering-and-pricing.writes.write-size-multiple-events)」で説明されているように、書き込みのコストをさらに最適化できます。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
public void writeRecordsWithCommonAttributes() {
System.out.println("Writing records with extracting common attributes");
// Specify repeated values for all records
List records = new ArrayList<>();
final long time = System.currentTimeMillis();
List dimensions = new ArrayList<>();
final Dimension region = new Dimension().withName("region").withValue("us-east-1");
final Dimension az = new Dimension().withName("az").withValue("az1");
final Dimension hostname = new Dimension().withName("hostname").withValue("host1");
dimensions.add(region);
dimensions.add(az);
dimensions.add(hostname);
Record commonAttributes = new Record()
.withDimensions(dimensions)
.withMeasureValueType(MeasureValueType.DOUBLE)
.withTime(String.valueOf(time));
Record cpuUtilization = new Record()
.withMeasureName("cpu_utilization")
.withMeasureValue("13.5");
Record memoryUtilization = new Record()
.withMeasureName("memory_utilization")
.withMeasureValue("40");
records.add(cpuUtilization);
records.add(memoryUtilization);
WriteRecordsRequest writeRecordsRequest = new WriteRecordsRequest()
.withDatabaseName(DATABASE_NAME)
.withTableName(TABLE_NAME)
.withCommonAttributes(commonAttributes);
writeRecordsRequest.setRecords(records);
try {
WriteRecordsResult writeRecordsResult = amazonTimestreamWrite.writeRecords(writeRecordsRequest);
System.out.println("writeRecordsWithCommonAttributes Status: " + writeRecordsResult.getSdkHttpMetadata().getHttpStatusCode());
} catch (RejectedRecordsException e) {
System.out.println("RejectedRecords: " + e);
for (RejectedRecord rejectedRecord : e.getRejectedRecords()) {
System.out.println("Rejected Index " + rejectedRecord.getRecordIndex() + ": "
+ rejectedRecord.getReason());
}
System.out.println("Other records were written successfully. ");
} catch (Exception e) {
System.out.println("Error: " + e);
}
}
```
------
#### [ Java v2 ]
```
public void writeRecordsWithCommonAttributes() {
System.out.println("Writing records with extracting common attributes");
// Specify repeated values for all records
List records = new ArrayList<>();
final long time = System.currentTimeMillis();
List dimensions = new ArrayList<>();
final Dimension region = Dimension.builder().name("region").value("us-east-1").build();
final Dimension az = Dimension.builder().name("az").value("az1").build();
final Dimension hostname = Dimension.builder().name("hostname").value("host1").build();
dimensions.add(region);
dimensions.add(az);
dimensions.add(hostname);
Record commonAttributes = Record.builder()
.dimensions(dimensions)
.measureValueType(MeasureValueType.DOUBLE)
.time(String.valueOf(time)).build();
Record cpuUtilization = Record.builder()
.measureName("cpu_utilization")
.measureValue("13.5").build();
Record memoryUtilization = Record.builder()
.measureName("memory_utilization")
.measureValue("40").build();
records.add(cpuUtilization);
records.add(memoryUtilization);
WriteRecordsRequest writeRecordsRequest = WriteRecordsRequest.builder()
.databaseName(DATABASE_NAME)
.tableName(TABLE_NAME)
.commonAttributes(commonAttributes)
.records(records).build();
try {
WriteRecordsResponse writeRecordsResponse = timestreamWriteClient.writeRecords(writeRecordsRequest);
System.out.println("writeRecordsWithCommonAttributes Status: " + writeRecordsResponse.sdkHttpResponse().statusCode());
} catch (RejectedRecordsException e) {
System.out.println("RejectedRecords: " + e);
for (RejectedRecord rejectedRecord : e.rejectedRecords()) {
System.out.println("Rejected Index " + rejectedRecord.recordIndex() + ": "
+ rejectedRecord.reason());
}
System.out.println("Other records were written successfully. ");
} catch (Exception e) {
System.out.println("Error: " + e);
}
}
```
------
#### [ Go ]
```
now = time.Now()
currentTimeInSeconds = now.Unix()
writeRecordsCommonAttributesInput := ×treamwrite.WriteRecordsInput{
DatabaseName: aws.String(*databaseName),
TableName: aws.String(*tableName),
CommonAttributes: ×treamwrite.Record{
Dimensions: []*timestreamwrite.Dimension{
×treamwrite.Dimension{
Name: aws.String("region"),
Value: aws.String("us-east-1"),
},
×treamwrite.Dimension{
Name: aws.String("az"),
Value: aws.String("az1"),
},
×treamwrite.Dimension{
Name: aws.String("hostname"),
Value: aws.String("host1"),
},
},
MeasureValueType: aws.String("DOUBLE"),
Time: aws.String(strconv.FormatInt(currentTimeInSeconds, 10)),
TimeUnit: aws.String("SECONDS"),
},
Records: []*timestreamwrite.Record{
×treamwrite.Record{
MeasureName: aws.String("cpu_utilization"),
MeasureValue: aws.String("13.5"),
},
×treamwrite.Record{
MeasureName: aws.String("memory_utilization"),
MeasureValue: aws.String("40"),
},
},
}
_, err = writeSvc.WriteRecords(writeRecordsCommonAttributesInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Ingest records is successful")
}
```
------
#### [ Python ]
```
def write_records_with_common_attributes(self):
print("Writing records extracting common attributes")
current_time = self._current_milli_time()
dimensions = [
{'Name': 'region', 'Value': 'us-east-1'},
{'Name': 'az', 'Value': 'az1'},
{'Name': 'hostname', 'Value': 'host1'}
]
common_attributes = {
'Dimensions': dimensions,
'MeasureValueType': 'DOUBLE',
'Time': current_time
}
cpu_utilization = {
'MeasureName': 'cpu_utilization',
'MeasureValue': '13.5'
}
memory_utilization = {
'MeasureName': 'memory_utilization',
'MeasureValue': '40'
}
records = [cpu_utilization, memory_utilization]
try:
result = self.client.write_records(DatabaseName=Constant.DATABASE_NAME, TableName=Constant.TABLE_NAME,
Records=records, CommonAttributes=common_attributes)
print("WriteRecords Status: [%s]" % result['ResponseMetadata']['HTTPStatusCode'])
except self.client.exceptions.RejectedRecordsException as err:
self._print_rejected_records_exceptions(err)
except Exception as err:
print("Error:", err)
@staticmethod
def _print_rejected_records_exceptions(err):
print("RejectedRecords: ", err)
for rr in err.response["RejectedRecords"]:
print("Rejected Index " + str(rr["RecordIndex"]) + ": " + rr["Reason"])
if "ExistingVersion" in rr:
print("Rejected record existing version: ", rr["ExistingVersion"])
@staticmethod
def _current_milli_time():
return str(int(round(time.time() * 1000)))
```
------
#### [ Node.js ]
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
async function writeRecordsWithCommonAttributes() {
console.log("Writing records with common attributes");
const currentTime = Date.now().toString(); // Unix time in milliseconds
const dimensions = [
{'Name': 'region', 'Value': 'us-east-1'},
{'Name': 'az', 'Value': 'az1'},
{'Name': 'hostname', 'Value': 'host1'}
];
const commonAttributes = {
'Dimensions': dimensions,
'MeasureValueType': 'DOUBLE',
'Time': currentTime.toString()
};
const cpuUtilization = {
'MeasureName': 'cpu_utilization',
'MeasureValue': '13.5'
};
const memoryUtilization = {
'MeasureName': 'memory_utilization',
'MeasureValue': '40'
};
const records = [cpuUtilization, memoryUtilization];
const params = {
DatabaseName: constants.DATABASE_NAME,
TableName: constants.TABLE_NAME,
Records: records,
CommonAttributes: commonAttributes
};
const request = writeClient.writeRecords(params);
await request.promise().then(
(data) => {
console.log("Write records successful");
},
(err) => {
console.log("Error writing records:", err);
if (err.code === 'RejectedRecordsException') {
const responsePayload = JSON.parse(request.response.httpResponse.body.toString());
console.log("RejectedRecords: ", responsePayload.RejectedRecords);
console.log("Other records were written successfully. ");
}
}
);
}
```
------
#### [ .NET ]
```
public async Task WriteRecordsWithCommonAttributes()
{
Console.WriteLine("Writing records with common attributes");
DateTimeOffset now = DateTimeOffset.UtcNow;
string currentTimeString = (now.ToUnixTimeMilliseconds()).ToString();
List dimensions = new List{
new Dimension { Name = "region", Value = "us-east-1" },
new Dimension { Name = "az", Value = "az1" },
new Dimension { Name = "hostname", Value = "host1" }
};
var commonAttributes = new Record
{
Dimensions = dimensions,
MeasureValueType = MeasureValueType.DOUBLE,
Time = currentTimeString
};
var cpuUtilization = new Record
{
MeasureName = "cpu_utilization",
MeasureValue = "13.6"
};
var memoryUtilization = new Record
{
MeasureName = "memory_utilization",
MeasureValue = "40"
};
List records = new List();
records.Add(cpuUtilization);
records.Add(memoryUtilization);
try
{
var writeRecordsRequest = new WriteRecordsRequest
{
DatabaseName = Constants.DATABASE_NAME,
TableName = Constants.TABLE_NAME,
Records = records,
CommonAttributes = commonAttributes
};
WriteRecordsResponse response = await writeClient.WriteRecordsAsync(writeRecordsRequest);
Console.WriteLine($"Write records status code: {response.HttpStatusCode.ToString()}");
}
catch (RejectedRecordsException e) {
Console.WriteLine("RejectedRecordsException:" + e.ToString());
foreach (RejectedRecord rr in e.RejectedRecords) {
Console.WriteLine("RecordIndex " + rr.RecordIndex + " : " + rr.Reason);
}
Console.WriteLine("Other records were written successfully. ");
}
catch (Exception e)
{
Console.WriteLine("Write records failure:" + e.ToString());
}
}
```
------
## レコードのアップサート
Amazon Timestream のデフォルトの書き込みは *first writer wins* セマンティクスに従い、データは追加専用として保存され、重複するレコードは拒否されますが、*last writer wins* セマンティクスを使用して Amazon Timestream にデータを書き込む必要があるアプリケーションがあり、その場合はバージョンが最も新しいレコードがシステムに保存されます。既存のレコードを更新する必要があるアプリケーションもあります。これらのシナリオに対応するため、Amazon Timestream はデータを*アップサート*する機能を提供します。アップサートは、レコードが存在しない場合はシステムにレコードを挿入し、存在する場合はレコードを更新するオペレーションです。
`WriteRecords` リクエストの送信中にレコード定義に `Version` を含めることで、レコードをアップサートできます。Amazon Timestream は、`Version` が最も新しいレコードと共にレコードを保存します。以下のコードサンプルは、データをアップサートする方法を示しています。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
public void writeRecordsWithUpsert() {
System.out.println("Writing records with upsert");
// Specify repeated values for all records
List records = new ArrayList<>();
final long time = System.currentTimeMillis();
// To achieve upsert (last writer wins) semantic, one example is to use current time as the version if you are writing directly from the data source
long version = System.currentTimeMillis();
List dimensions = new ArrayList<>();
final Dimension region = new Dimension().withName("region").withValue("us-east-1");
final Dimension az = new Dimension().withName("az").withValue("az1");
final Dimension hostname = new Dimension().withName("hostname").withValue("host1");
dimensions.add(region);
dimensions.add(az);
dimensions.add(hostname);
Record commonAttributes = new Record()
.withDimensions(dimensions)
.withMeasureValueType(MeasureValueType.DOUBLE)
.withTime(String.valueOf(time))
.withVersion(version);
Record cpuUtilization = new Record()
.withMeasureName("cpu_utilization")
.withMeasureValue("13.5");
Record memoryUtilization = new Record()
.withMeasureName("memory_utilization")
.withMeasureValue("40");
records.add(cpuUtilization);
records.add(memoryUtilization);
WriteRecordsRequest writeRecordsRequest = new WriteRecordsRequest()
.withDatabaseName(DATABASE_NAME)
.withTableName(TABLE_NAME)
.withCommonAttributes(commonAttributes);
writeRecordsRequest.setRecords(records);
// write records for first time
try {
WriteRecordsResult writeRecordsResult = amazonTimestreamWrite.writeRecords(writeRecordsRequest);
System.out.println("WriteRecords Status for first time: " + writeRecordsResult.getSdkHttpMetadata().getHttpStatusCode());
} catch (RejectedRecordsException e) {
printRejectedRecordsException(e);
} catch (Exception e) {
System.out.println("Error: " + e);
}
// Successfully retry same writeRecordsRequest with same records and versions, because writeRecords API is idempotent.
try {
WriteRecordsResult writeRecordsResult = amazonTimestreamWrite.writeRecords(writeRecordsRequest);
System.out.println("WriteRecords Status for retry: " + writeRecordsResult.getSdkHttpMetadata().getHttpStatusCode());
} catch (RejectedRecordsException e) {
printRejectedRecordsException(e);
} catch (Exception e) {
System.out.println("Error: " + e);
}
// upsert with lower version, this would fail because a higher version is required to update the measure value.
version -= 1;
commonAttributes.setVersion(version);
cpuUtilization.setMeasureValue("14.5");
memoryUtilization.setMeasureValue("50");
List upsertedRecords = new ArrayList<>();
upsertedRecords.add(cpuUtilization);
upsertedRecords.add(memoryUtilization);
WriteRecordsRequest writeRecordsUpsertRequest = new WriteRecordsRequest()
.withDatabaseName(DATABASE_NAME)
.withTableName(TABLE_NAME)
.withCommonAttributes(commonAttributes);
writeRecordsUpsertRequest.setRecords(upsertedRecords);
try {
WriteRecordsResult writeRecordsUpsertResult = amazonTimestreamWrite.writeRecords(writeRecordsUpsertRequest);
System.out.println("WriteRecords Status for upsert with lower version: " + writeRecordsUpsertResult.getSdkHttpMetadata().getHttpStatusCode());
} catch (RejectedRecordsException e) {
System.out.println("WriteRecords Status for upsert with lower version: ");
printRejectedRecordsException(e);
} catch (Exception e) {
System.out.println("Error: " + e);
}
// upsert with higher version as new data in generated
version = System.currentTimeMillis();
commonAttributes.setVersion(version);
writeRecordsUpsertRequest = new WriteRecordsRequest()
.withDatabaseName(DATABASE_NAME)
.withTableName(TABLE_NAME)
.withCommonAttributes(commonAttributes);
writeRecordsUpsertRequest.setRecords(upsertedRecords);
try {
WriteRecordsResult writeRecordsUpsertResult = amazonTimestreamWrite.writeRecords(writeRecordsUpsertRequest);
System.out.println("WriteRecords Status for upsert with higher version: " + writeRecordsUpsertResult.getSdkHttpMetadata().getHttpStatusCode());
} catch (RejectedRecordsException e) {
printRejectedRecordsException(e);
} catch (Exception e) {
System.out.println("Error: " + e);
}
}
```
------
#### [ Java v2 ]
```
public void writeRecordsWithUpsert() {
System.out.println("Writing records with upsert");
// Specify repeated values for all records
List records = new ArrayList<>();
final long time = System.currentTimeMillis();
// To achieve upsert (last writer wins) semantic, one example is to use current time as the version if you are writing directly from the data source
long version = System.currentTimeMillis();
List dimensions = new ArrayList<>();
final Dimension region = Dimension.builder().name("region").value("us-east-1").build();
final Dimension az = Dimension.builder().name("az").value("az1").build();
final Dimension hostname = Dimension.builder().name("hostname").value("host1").build();
dimensions.add(region);
dimensions.add(az);
dimensions.add(hostname);
Record commonAttributes = Record.builder()
.dimensions(dimensions)
.measureValueType(MeasureValueType.DOUBLE)
.time(String.valueOf(time))
.version(version)
.build();
Record cpuUtilization = Record.builder()
.measureName("cpu_utilization")
.measureValue("13.5").build();
Record memoryUtilization = Record.builder()
.measureName("memory_utilization")
.measureValue("40").build();
records.add(cpuUtilization);
records.add(memoryUtilization);
WriteRecordsRequest writeRecordsRequest = WriteRecordsRequest.builder()
.databaseName(DATABASE_NAME)
.tableName(TABLE_NAME)
.commonAttributes(commonAttributes)
.records(records).build();
// write records for first time
try {
WriteRecordsResponse writeRecordsResponse = timestreamWriteClient.writeRecords(writeRecordsRequest);
System.out.println("WriteRecords Status for first time: " + writeRecordsResponse.sdkHttpResponse().statusCode());
} catch (RejectedRecordsException e) {
printRejectedRecordsException(e);
} catch (Exception e) {
System.out.println("Error: " + e);
}
// Successfully retry same writeRecordsRequest with same records and versions, because writeRecords API is idempotent.
try {
WriteRecordsResponse writeRecordsResponse = timestreamWriteClient.writeRecords(writeRecordsRequest);
System.out.println("WriteRecords Status for retry: " + writeRecordsResponse.sdkHttpResponse().statusCode());
} catch (RejectedRecordsException e) {
printRejectedRecordsException(e);
} catch (Exception e) {
System.out.println("Error: " + e);
}
// upsert with lower version, this would fail because a higher version is required to update the measure value.
version -= 1;
commonAttributes = Record.builder()
.dimensions(dimensions)
.measureValueType(MeasureValueType.DOUBLE)
.time(String.valueOf(time))
.version(version)
.build();
cpuUtilization = Record.builder()
.measureName("cpu_utilization")
.measureValue("14.5").build();
memoryUtilization = Record.builder()
.measureName("memory_utilization")
.measureValue("50").build();
List upsertedRecords = new ArrayList<>();
upsertedRecords.add(cpuUtilization);
upsertedRecords.add(memoryUtilization);
WriteRecordsRequest writeRecordsUpsertRequest = WriteRecordsRequest.builder()
.databaseName(DATABASE_NAME)
.tableName(TABLE_NAME)
.commonAttributes(commonAttributes)
.records(upsertedRecords).build();
try {
WriteRecordsResponse writeRecordsResponse = timestreamWriteClient.writeRecords(writeRecordsUpsertRequest);
System.out.println("WriteRecords Status for upsert with lower version: " + writeRecordsResponse.sdkHttpResponse().statusCode());
} catch (RejectedRecordsException e) {
System.out.println("WriteRecords Status for upsert with lower version: ");
printRejectedRecordsException(e);
} catch (Exception e) {
System.out.println("Error: " + e);
}
// upsert with higher version as new data in generated
version = System.currentTimeMillis();
commonAttributes = Record.builder()
.dimensions(dimensions)
.measureValueType(MeasureValueType.DOUBLE)
.time(String.valueOf(time))
.version(version)
.build();
writeRecordsUpsertRequest = WriteRecordsRequest.builder()
.databaseName(DATABASE_NAME)
.tableName(TABLE_NAME)
.commonAttributes(commonAttributes)
.records(upsertedRecords).build();
try {
WriteRecordsResponse writeRecordsUpsertResponse = timestreamWriteClient.writeRecords(writeRecordsUpsertRequest);
System.out.println("WriteRecords Status for upsert with higher version: " + writeRecordsUpsertResponse.sdkHttpResponse().statusCode());
} catch (RejectedRecordsException e) {
printRejectedRecordsException(e);
} catch (Exception e) {
System.out.println("Error: " + e);
}
}
```
------
#### [ Go ]
```
// Below code will ingest and upsert cpu_utilization and memory_utilization metric for a host on
// region=us-east-1, az=az1, and hostname=host1
fmt.Println("Ingesting records and set version as currentTimeInMills, hit enter to continue")
reader.ReadString('\n')
// Get current time in seconds.
now = time.Now()
currentTimeInSeconds = now.Unix()
// To achieve upsert (last writer wins) semantic, one example is to use current time as the version if you are writing directly from the data source
version := time.Now().Round(time.Millisecond).UnixNano() / 1e6 // set version as currentTimeInMills
writeRecordsCommonAttributesUpsertInput := ×treamwrite.WriteRecordsInput{
DatabaseName: aws.String(*databaseName),
TableName: aws.String(*tableName),
CommonAttributes: ×treamwrite.Record{
Dimensions: []*timestreamwrite.Dimension{
×treamwrite.Dimension{
Name: aws.String("region"),
Value: aws.String("us-east-1"),
},
×treamwrite.Dimension{
Name: aws.String("az"),
Value: aws.String("az1"),
},
×treamwrite.Dimension{
Name: aws.String("hostname"),
Value: aws.String("host1"),
},
},
MeasureValueType: aws.String("DOUBLE"),
Time: aws.String(strconv.FormatInt(currentTimeInSeconds, 10)),
TimeUnit: aws.String("SECONDS"),
Version: &version,
},
Records: []*timestreamwrite.Record{
×treamwrite.Record{
MeasureName: aws.String("cpu_utilization"),
MeasureValue: aws.String("13.5"),
},
×treamwrite.Record{
MeasureName: aws.String("memory_utilization"),
MeasureValue: aws.String("40"),
},
},
}
// write records for first time
_, err = writeSvc.WriteRecords(writeRecordsCommonAttributesUpsertInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Frist-time write records is successful")
}
fmt.Println("Retry same writeRecordsRequest with same records and versions. Because writeRecords API is idempotent, this will success. hit enter to continue")
reader.ReadString('\n')
_, err = writeSvc.WriteRecords(writeRecordsCommonAttributesUpsertInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Retry write records for same request is successful")
}
fmt.Println("Upsert with lower version, this would fail because a higher version is required to update the measure value. hit enter to continue")
reader.ReadString('\n')
version -= 1
writeRecordsCommonAttributesUpsertInput.CommonAttributes.Version = &version
updated_cpu_utilization := ×treamwrite.Record{
MeasureName: aws.String("cpu_utilization"),
MeasureValue: aws.String("14.5"),
}
updated_memory_utilization := ×treamwrite.Record{
MeasureName: aws.String("memory_utilization"),
MeasureValue: aws.String("50"),
}
writeRecordsCommonAttributesUpsertInput.Records = []*timestreamwrite.Record{
updated_cpu_utilization,
updated_memory_utilization,
}
_, err = writeSvc.WriteRecords(writeRecordsCommonAttributesUpsertInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Write records with lower version is successful")
}
fmt.Println("Upsert with higher version as new data in generated, this would success. hit enter to continue")
reader.ReadString('\n')
version = time.Now().Round(time.Millisecond).UnixNano() / 1e6 // set version as currentTimeInMills
writeRecordsCommonAttributesUpsertInput.CommonAttributes.Version = &version
_, err = writeSvc.WriteRecords(writeRecordsCommonAttributesUpsertInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Write records with higher version is successful")
}
```
------
#### [ Python ]
```
def write_records_with_upsert(self):
print("Writing records with upsert")
current_time = self._current_milli_time()
# To achieve upsert (last writer wins) semantic, one example is to use current time as the version if you are writing directly from the data source
version = int(self._current_milli_time())
dimensions = [
{'Name': 'region', 'Value': 'us-east-1'},
{'Name': 'az', 'Value': 'az1'},
{'Name': 'hostname', 'Value': 'host1'}
]
common_attributes = {
'Dimensions': dimensions,
'MeasureValueType': 'DOUBLE',
'Time': current_time,
'Version': version
}
cpu_utilization = {
'MeasureName': 'cpu_utilization',
'MeasureValue': '13.5'
}
memory_utilization = {
'MeasureName': 'memory_utilization',
'MeasureValue': '40'
}
records = [cpu_utilization, memory_utilization]
# write records for first time
try:
result = self.client.write_records(DatabaseName=Constant.DATABASE_NAME, TableName=Constant.TABLE_NAME,
Records=records, CommonAttributes=common_attributes)
print("WriteRecords Status for first time: [%s]" % result['ResponseMetadata']['HTTPStatusCode'])
except self.client.exceptions.RejectedRecordsException as err:
self._print_rejected_records_exceptions(err)
except Exception as err:
print("Error:", err)
# Successfully retry same writeRecordsRequest with same records and versions, because writeRecords API is idempotent.
try:
result = self.client.write_records(DatabaseName=Constant.DATABASE_NAME, TableName=Constant.TABLE_NAME,
Records=records, CommonAttributes=common_attributes)
print("WriteRecords Status for retry: [%s]" % result['ResponseMetadata']['HTTPStatusCode'])
except self.client.exceptions.RejectedRecordsException as err:
self._print_rejected_records_exceptions(err)
except Exception as err:
print("Error:", err)
# upsert with lower version, this would fail because a higher version is required to update the measure value.
version -= 1
common_attributes["Version"] = version
cpu_utilization["MeasureValue"] = '14.5'
memory_utilization["MeasureValue"] = '50'
upsertedRecords = [cpu_utilization, memory_utilization]
try:
upsertedResult = self.client.write_records(DatabaseName=Constant.DATABASE_NAME, TableName=Constant.TABLE_NAME,
Records=upsertedRecords, CommonAttributes=common_attributes)
print("WriteRecords Status for upsert with lower version: [%s]" % upsertedResult['ResponseMetadata']['HTTPStatusCode'])
except self.client.exceptions.RejectedRecordsException as err:
self._print_rejected_records_exceptions(err)
except Exception as err:
print("Error:", err)
# upsert with higher version as new data is generated
version = int(self._current_milli_time())
common_attributes["Version"] = version
try:
upsertedResult = self.client.write_records(DatabaseName=Constant.DATABASE_NAME, TableName=Constant.TABLE_NAME,
Records=upsertedRecords, CommonAttributes=common_attributes)
print("WriteRecords Upsert Status: [%s]" % upsertedResult['ResponseMetadata']['HTTPStatusCode'])
except self.client.exceptions.RejectedRecordsException as err:
self._print_rejected_records_exceptions(err)
except Exception as err:
print("Error:", err)
@staticmethod
def _current_milli_time():
return str(int(round(time.time() * 1000)))
```
------
#### [ Node.js ]
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
async function writeRecordsWithUpsert() {
console.log("Writing records with upsert");
const currentTime = Date.now().toString(); // Unix time in milliseconds
// To achieve upsert (last writer wins) semantic, one example is to use current time as the version if you are writing directly from the data source
let version = Date.now();
const dimensions = [
{'Name': 'region', 'Value': 'us-east-1'},
{'Name': 'az', 'Value': 'az1'},
{'Name': 'hostname', 'Value': 'host1'}
];
const commonAttributes = {
'Dimensions': dimensions,
'MeasureValueType': 'DOUBLE',
'Time': currentTime.toString(),
'Version': version
};
const cpuUtilization = {
'MeasureName': 'cpu_utilization',
'MeasureValue': '13.5'
};
const memoryUtilization = {
'MeasureName': 'memory_utilization',
'MeasureValue': '40'
};
const records = [cpuUtilization, memoryUtilization];
const params = {
DatabaseName: constants.DATABASE_NAME,
TableName: constants.TABLE_NAME,
Records: records,
CommonAttributes: commonAttributes
};
const request = writeClient.writeRecords(params);
// write records for first time
await request.promise().then(
(data) => {
console.log("Write records successful for first time.");
},
(err) => {
console.log("Error writing records:", err);
if (err.code === 'RejectedRecordsException') {
printRejectedRecordsException(request);
}
}
);
// Successfully retry same writeRecordsRequest with same records and versions, because writeRecords API is idempotent.
await request.promise().then(
(data) => {
console.log("Write records successful for retry.");
},
(err) => {
console.log("Error writing records:", err);
if (err.code === 'RejectedRecordsException') {
printRejectedRecordsException(request);
}
}
);
// upsert with lower version, this would fail because a higher version is required to update the measure value.
version--;
const commonAttributesWithLowerVersion = {
'Dimensions': dimensions,
'MeasureValueType': 'DOUBLE',
'Time': currentTime.toString(),
'Version': version
};
const updatedCpuUtilization = {
'MeasureName': 'cpu_utilization',
'MeasureValue': '14.5'
};
const updatedMemoryUtilization = {
'MeasureName': 'memory_utilization',
'MeasureValue': '50'
};
const upsertedRecords = [updatedCpuUtilization, updatedMemoryUtilization];
const upsertedParamsWithLowerVersion = {
DatabaseName: constants.DATABASE_NAME,
TableName: constants.TABLE_NAME,
Records: upsertedRecords,
CommonAttributes: commonAttributesWithLowerVersion
};
const upsertRequestWithLowerVersion = writeClient.writeRecords(upsertedParamsWithLowerVersion);
await upsertRequestWithLowerVersion.promise().then(
(data) => {
console.log("Write records for upsert with lower version successful");
},
(err) => {
console.log("Error writing records:", err);
if (err.code === 'RejectedRecordsException') {
printRejectedRecordsException(upsertRequestWithLowerVersion);
}
}
);
// upsert with higher version as new data in generated
version = Date.now();
const commonAttributesWithHigherVersion = {
'Dimensions': dimensions,
'MeasureValueType': 'DOUBLE',
'Time': currentTime.toString(),
'Version': version
};
const upsertedParamsWithHigherVerion = {
DatabaseName: constants.DATABASE_NAME,
TableName: constants.TABLE_NAME,
Records: upsertedRecords,
CommonAttributes: commonAttributesWithHigherVersion
};
const upsertRequestWithHigherVersion = writeClient.writeRecords(upsertedParamsWithHigherVerion);
await upsertRequestWithHigherVersion.promise().then(
(data) => {
console.log("Write records upsert successful with higher version");
},
(err) => {
console.log("Error writing records:", err);
if (err.code === 'RejectedRecordsException') {
printRejectedRecordsException(upsertedParamsWithHigherVerion);
}
}
);
}
```
------
#### [ .NET ]
```
public async Task WriteRecordsWithUpsert()
{
Console.WriteLine("Writing records with upsert");
DateTimeOffset now = DateTimeOffset.UtcNow;
string currentTimeString = (now.ToUnixTimeMilliseconds()).ToString();
// To achieve upsert (last writer wins) semantic, one example is to use current time as the version if you are writing directly from the data source
long version = now.ToUnixTimeMilliseconds();
List dimensions = new List{
new Dimension { Name = "region", Value = "us-east-1" },
new Dimension { Name = "az", Value = "az1" },
new Dimension { Name = "hostname", Value = "host1" }
};
var commonAttributes = new Record
{
Dimensions = dimensions,
MeasureValueType = MeasureValueType.DOUBLE,
Time = currentTimeString,
Version = version
};
var cpuUtilization = new Record
{
MeasureName = "cpu_utilization",
MeasureValue = "13.6"
};
var memoryUtilization = new Record
{
MeasureName = "memory_utilization",
MeasureValue = "40"
};
List records = new List();
records.Add(cpuUtilization);
records.Add(memoryUtilization);
// write records for first time
try
{
var writeRecordsRequest = new WriteRecordsRequest
{
DatabaseName = Constants.DATABASE_NAME,
TableName = Constants.TABLE_NAME,
Records = records,
CommonAttributes = commonAttributes
};
WriteRecordsResponse response = await writeClient.WriteRecordsAsync(writeRecordsRequest);
Console.WriteLine($"WriteRecords Status for first time: {response.HttpStatusCode.ToString()}");
}
catch (RejectedRecordsException e) {
PrintRejectedRecordsException(e);
}
catch (Exception e)
{
Console.WriteLine("Write records failure:" + e.ToString());
}
// Successfully retry same writeRecordsRequest with same records and versions, because writeRecords API is idempotent.
try
{
var writeRecordsRequest = new WriteRecordsRequest
{
DatabaseName = Constants.DATABASE_NAME,
TableName = Constants.TABLE_NAME,
Records = records,
CommonAttributes = commonAttributes
};
WriteRecordsResponse response = await writeClient.WriteRecordsAsync(writeRecordsRequest);
Console.WriteLine($"WriteRecords Status for retry: {response.HttpStatusCode.ToString()}");
}
catch (RejectedRecordsException e) {
PrintRejectedRecordsException(e);
}
catch (Exception e)
{
Console.WriteLine("Write records failure:" + e.ToString());
}
// upsert with lower version, this would fail because a higher version is required to update the measure value.
version--;
Type recordType = typeof(Record);
recordType.GetProperty("Version").SetValue(commonAttributes, version);
recordType.GetProperty("MeasureValue").SetValue(cpuUtilization, "14.6");
recordType.GetProperty("MeasureValue").SetValue(memoryUtilization, "50");
List upsertedRecords = new List {
cpuUtilization,
memoryUtilization
};
try
{
var writeRecordsUpsertRequest = new WriteRecordsRequest
{
DatabaseName = Constants.DATABASE_NAME,
TableName = Constants.TABLE_NAME,
Records = upsertedRecords,
CommonAttributes = commonAttributes
};
WriteRecordsResponse upsertResponse = await writeClient.WriteRecordsAsync(writeRecordsUpsertRequest);
Console.WriteLine($"WriteRecords Status for upsert with lower version: {upsertResponse.HttpStatusCode.ToString()}");
}
catch (RejectedRecordsException e) {
PrintRejectedRecordsException(e);
}
catch (Exception e)
{
Console.WriteLine("Write records failure:" + e.ToString());
}
// upsert with higher version as new data in generated
now = DateTimeOffset.UtcNow;
version = now.ToUnixTimeMilliseconds();
recordType.GetProperty("Version").SetValue(commonAttributes, version);
try
{
var writeRecordsUpsertRequest = new WriteRecordsRequest
{
DatabaseName = Constants.DATABASE_NAME,
TableName = Constants.TABLE_NAME,
Records = upsertedRecords,
CommonAttributes = commonAttributes
};
WriteRecordsResponse upsertResponse = await writeClient.WriteRecordsAsync(writeRecordsUpsertRequest);
Console.WriteLine($"WriteRecords Status for upsert with higher version: {upsertResponse.HttpStatusCode.ToString()}");
}
catch (RejectedRecordsException e) {
PrintRejectedRecordsException(e);
}
catch (Exception e)
{
Console.WriteLine("Write records failure:" + e.ToString());
}
}
```
------
## マルチメジャー属性の例
この例では、マルチメジャー属性の記述を示しています。[マルチメジャー属性](data-modeling.md#data-modeling-multiVsinglerecords)は、追跡しているデバイスまたはアプリケーションが複数のメトリクスまたはイベントを同じタイムスタンプで出力する場合に便利です。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
package com.amazonaws.services.timestream;
import static com.amazonaws.services.timestream.Main.DATABASE_NAME;
import static com.amazonaws.services.timestream.Main.REGION;
import static com.amazonaws.services.timestream.Main.TABLE_NAME;
import java.util.ArrayList;
import java.util.List;
import com.amazonaws.services.timestreamwrite.AmazonTimestreamWrite;
import com.amazonaws.services.timestreamwrite.model.Dimension;
import com.amazonaws.services.timestreamwrite.model.MeasureValue;
import com.amazonaws.services.timestreamwrite.model.MeasureValueType;
import com.amazonaws.services.timestreamwrite.model.Record;
import com.amazonaws.services.timestreamwrite.model.RejectedRecordsException;
import com.amazonaws.services.timestreamwrite.model.WriteRecordsRequest;
import com.amazonaws.services.timestreamwrite.model.WriteRecordsResult;
public class multimeasureAttributeExample {
AmazonTimestreamWrite timestreamWriteClient;
public multimeasureAttributeExample(AmazonTimestreamWrite client) {
this.timestreamWriteClient = client;
}
public void writeRecordsMultiMeasureValueSingleRecord() {
System.out.println("Writing records with multi value attributes");
List records = new ArrayList<>();
final long time = System.currentTimeMillis();
long version = System.currentTimeMillis();
List dimensions = new ArrayList<>();
final Dimension region = new Dimension().withName("region").withValue(REGION);
final Dimension az = new Dimension().withName("az").withValue("az1");
final Dimension hostname = new Dimension().withName("hostname").withValue("host1");
dimensions.add(region);
dimensions.add(az);
dimensions.add(hostname);
Record commonAttributes = new Record()
.withDimensions(dimensions)
.withTime(String.valueOf(time))
.withVersion(version);
MeasureValue cpuUtilization = new MeasureValue()
.withName("cpu_utilization")
.withType(MeasureValueType.DOUBLE)
.withValue("13.5");
MeasureValue memoryUtilization = new MeasureValue()
.withName("memory_utilization")
.withType(MeasureValueType.DOUBLE)
.withValue("40");
Record computationalResources = new Record()
.withMeasureName("cpu_memory")
.withMeasureValues(cpuUtilization, memoryUtilization)
.withMeasureValueType(MeasureValueType.MULTI);
records.add(computationalResources);
WriteRecordsRequest writeRecordsRequest = new WriteRecordsRequest()
.withDatabaseName(DATABASE_NAME)
.withTableName(TABLE_NAME)
.withCommonAttributes(commonAttributes)
.withRecords(records);
// write records for first time
try {
WriteRecordsResult writeRecordResult = timestreamWriteClient.writeRecords(writeRecordsRequest);
System.out.println(
"WriteRecords Status for multi value attributes: " + writeRecordResult
.getSdkHttpMetadata().getHttpStatusCode());
} catch (RejectedRecordsException e) {
printRejectedRecordsException(e);
} catch (Exception e) {
System.out.println("Error: " + e);
}
}
public void writeRecordsMultiMeasureValueMultipleRecords() {
System.out.println(
"Writing records with multi value attributes mixture type");
List records = new ArrayList<>();
final long time = System.currentTimeMillis();
long version = System.currentTimeMillis();
List dimensions = new ArrayList<>();
final Dimension region = new Dimension().withName("region").withValue(REGION);
final Dimension az = new Dimension().withName("az").withValue("az1");
final Dimension hostname = new Dimension().withName("hostname").withValue("host1");
dimensions.add(region);
dimensions.add(az);
dimensions.add(hostname);
Record commonAttributes = new Record()
.withDimensions(dimensions)
.withTime(String.valueOf(time))
.withVersion(version);
MeasureValue cpuUtilization = new MeasureValue()
.withName("cpu_utilization")
.withType(MeasureValueType.DOUBLE)
.withValue("13");
MeasureValue memoryUtilization =new MeasureValue()
.withName("memory_utilization")
.withType(MeasureValueType.DOUBLE)
.withValue("40");
MeasureValue activeCores = new MeasureValue()
.withName("active_cores")
.withType(MeasureValueType.BIGINT)
.withValue("4");
Record computationalResources = new Record()
.withMeasureName("computational_utilization")
.withMeasureValues(cpuUtilization, memoryUtilization, activeCores)
.withMeasureValueType(MeasureValueType.MULTI);
records.add(computationalResources);
WriteRecordsRequest writeRecordsRequest = new WriteRecordsRequest()
.withDatabaseName(DATABASE_NAME)
.withTableName(TABLE_NAME)
.withCommonAttributes(commonAttributes)
.withRecords(records);
// write records for first time
try {
WriteRecordsResult writeRecordResult = timestreamWriteClient.writeRecords(writeRecordsRequest);
System.out.println(
"WriteRecords Status for multi value attributes: " + writeRecordResult
.getSdkHttpMetadata().getHttpStatusCode());
} catch (RejectedRecordsException e) {
printRejectedRecordsException(e);
} catch (Exception e) {
System.out.println("Error: " + e);
}
}
private void printRejectedRecordsException(RejectedRecordsException e) {
System.out.println("RejectedRecords: " + e);
e.getRejectedRecords().forEach(System.out::println);
}
}
```
------
#### [ Java v2 ]
```
package com.amazonaws.services.timestream;
import java.util.ArrayList;
import java.util.List;
import software.amazon.awssdk.services.timestreamwrite.TimestreamWriteClient;
import software.amazon.awssdk.services.timestreamwrite.model.Dimension;
import software.amazon.awssdk.services.timestreamwrite.model.MeasureValue;
import software.amazon.awssdk.services.timestreamwrite.model.MeasureValueType;
import software.amazon.awssdk.services.timestreamwrite.model.Record;
import software.amazon.awssdk.services.timestreamwrite.model.RejectedRecordsException;
import software.amazon.awssdk.services.timestreamwrite.model.WriteRecordsRequest;
import software.amazon.awssdk.services.timestreamwrite.model.WriteRecordsResponse;
import static com.amazonaws.services.timestream.Main.DATABASE_NAME;
import static com.amazonaws.services.timestream.Main.TABLE_NAME;
public class multimeasureAttributeExample {
TimestreamWriteClient timestreamWriteClient;
public multimeasureAttributeExample(TimestreamWriteClient client) {
this.timestreamWriteClient = client;
}
public void writeRecordsMultiMeasureValueSingleRecord() {
System.out.println("Writing records with multi value attributes");
List records = new ArrayList<>();
final long time = System.currentTimeMillis();
long version = System.currentTimeMillis();
List dimensions = new ArrayList<>();
final Dimension region =
Dimension.builder().name("region").value("us-east-1").build();
final Dimension az = Dimension.builder().name("az").value("az1").build();
final Dimension hostname =
Dimension.builder().name("hostname").value("host1").build();
dimensions.add(region);
dimensions.add(az);
dimensions.add(hostname);
Record commonAttributes = Record.builder()
.dimensions(dimensions)
.time(String.valueOf(time))
.version(version)
.build();
MeasureValue cpuUtilization = MeasureValue.builder()
.name("cpu_utilization")
.type(MeasureValueType.DOUBLE)
.value("13.5").build();
MeasureValue memoryUtilization = MeasureValue.builder()
.name("memory_utilization")
.type(MeasureValueType.DOUBLE)
.value("40").build();
Record computationalResources = Record
.builder()
.measureName("cpu_memory")
.measureValues(cpuUtilization, memoryUtilization)
.measureValueType(MeasureValueType.MULTI)
.build();
records.add(computationalResources);
WriteRecordsRequest writeRecordsRequest = WriteRecordsRequest.builder()
.databaseName(DATABASE_NAME)
.tableName(TABLE_NAME)
.commonAttributes(commonAttributes)
.records(records).build();
// write records for first time
try {
WriteRecordsResponse writeRecordsResponse = timestreamWriteClient.writeRecords(writeRecordsRequest);
System.out.println(
"WriteRecords Status for multi value attributes: " + writeRecordsResponse
.sdkHttpResponse()
.statusCode());
} catch (RejectedRecordsException e) {
printRejectedRecordsException(e);
} catch (Exception e) {
System.out.println("Error: " + e);
}
}
public void writeRecordsMultiMeasureValueMultipleRecords() {
System.out.println(
"Writing records with multi value attributes mixture type");
List records = new ArrayList<>();
final long time = System.currentTimeMillis();
long version = System.currentTimeMillis();
List dimensions = new ArrayList<>();
final Dimension region =
Dimension.builder().name("region").value("us-east-1").build();
final Dimension az = Dimension.builder().name("az").value("az1").build();
final Dimension hostname =
Dimension.builder().name("hostname").value("host1").build();
dimensions.add(region);
dimensions.add(az);
dimensions.add(hostname);
Record commonAttributes = Record.builder()
.dimensions(dimensions)
.time(String.valueOf(time))
.version(version)
.build();
MeasureValue cpuUtilization = MeasureValue.builder()
.name("cpu_utilization")
.type(MeasureValueType.DOUBLE)
.value("13.5").build();
MeasureValue memoryUtilization = MeasureValue.builder()
.name("memory_utilization")
.type(MeasureValueType.DOUBLE)
.value("40").build();
MeasureValue activeCores = MeasureValue.builder()
.name("active_cores")
.type(MeasureValueType.BIGINT)
.value("4").build();
Record computationalResources = Record
.builder()
.measureName("computational_utilization")
.measureValues(cpuUtilization, memoryUtilization, activeCores)
.measureValueType(MeasureValueType.MULTI)
.build();
records.add(computationalResources);
WriteRecordsRequest writeRecordsRequest = WriteRecordsRequest.builder()
.databaseName(DATABASE_NAME)
.tableName(TABLE_NAME)
.commonAttributes(commonAttributes)
.records(records).build();
// write records for first time
try {
WriteRecordsResponse writeRecordsResponse = timestreamWriteClient.writeRecords(writeRecordsRequest);
System.out.println(
"WriteRecords Status for multi value attributes: " + writeRecordsResponse
.sdkHttpResponse()
.statusCode());
} catch (RejectedRecordsException e) {
printRejectedRecordsException(e);
} catch (Exception e) {
System.out.println("Error: " + e);
}
}
private void printRejectedRecordsException(RejectedRecordsException e) {
System.out.println("RejectedRecords: " + e);
e.rejectedRecords().forEach(System.out::println);
}
}
```
------
#### [ Go ]
```
now := time.Now()
currentTimeInSeconds := now.Unix()
writeRecordsInput := ×treamwrite.WriteRecordsInput{
DatabaseName: aws.String(*databaseName),
TableName: aws.String(*tableName),
Records: []*timestreamwrite.Record{
×treamwrite.Record{
Dimensions: []*timestreamwrite.Dimension{
×treamwrite.Dimension{
Name: aws.String("region"),
Value: aws.String("us-east-1"),
},
×treamwrite.Dimension{
Name: aws.String("az"),
Value: aws.String("az1"),
},
×treamwrite.Dimension{
Name: aws.String("hostname"),
Value: aws.String("host1"),
},
},
MeasureName: aws.String("metrics"),
MeasureValueType: aws.String("MULTI"),
Time: aws.String(strconv.FormatInt(currentTimeInSeconds, 10)),
TimeUnit: aws.String("SECONDS"),
MeasureValues: []*timestreamwrite.MeasureValue{
×treamwrite.MeasureValue{
Name: aws.String("cpu_utilization"),
Value: aws.String("13.5"),
Type: aws.String("DOUBLE"),
},
×treamwrite.MeasureValue{
Name: aws.String("memory_utilization"),
Value: aws.String("40"),
Type: aws.String("DOUBLE"),
},
},
},
},
}
_, err = writeSvc.WriteRecords(writeRecordsInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Write records is successful")
}
```
------
#### [ Python ]
```
import time
import boto3
import psutil
import os
from botocore.config import Config
DATABASE_NAME = os.environ['DATABASE_NAME']
TABLE_NAME = os.environ['TABLE_NAME']
COUNTRY = "UK"
CITY = "London"
HOSTNAME = "MyHostname" # You can make it dynamic using socket.gethostname()
INTERVAL = 1 # Seconds
def prepare_common_attributes():
common_attributes = {
'Dimensions': [
{'Name': 'country', 'Value': COUNTRY},
{'Name': 'city', 'Value': CITY},
{'Name': 'hostname', 'Value': HOSTNAME}
],
'MeasureName': 'utilization',
'MeasureValueType': 'MULTI'
}
return common_attributes
def prepare_record(current_time):
record = {
'Time': str(current_time),
'MeasureValues': []
}
return record
def prepare_measure(measure_name, measure_value):
measure = {
'Name': measure_name,
'Value': str(measure_value),
'Type': 'DOUBLE'
}
return measure
def write_records(records, common_attributes):
try:
result = write_client.write_records(DatabaseName=DATABASE_NAME,
TableName=TABLE_NAME,
CommonAttributes=common_attributes,
Records=records)
status = result['ResponseMetadata']['HTTPStatusCode']
print("Processed %d records. WriteRecords HTTPStatusCode: %s" %
(len(records), status))
except Exception as err:
print("Error:", err)
if __name__ == '__main__':
print("writing data to database {} table {}".format(
DATABASE_NAME, TABLE_NAME))
session = boto3.Session()
write_client = session.client('timestream-write', config=Config(
read_timeout=20, max_pool_connections=5000, retries={'max_attempts': 10}))
query_client = session.client('timestream-query') # Not used
common_attributes = prepare_common_attributes()
records = []
while True:
current_time = int(time.time() * 1000)
cpu_utilization = psutil.cpu_percent()
memory_utilization = psutil.virtual_memory().percent
swap_utilization = psutil.swap_memory().percent
disk_utilization = psutil.disk_usage('/').percent
record = prepare_record(current_time)
record['MeasureValues'].append(prepare_measure('cpu', cpu_utilization))
record['MeasureValues'].append(prepare_measure('memory', memory_utilization))
record['MeasureValues'].append(prepare_measure('swap', swap_utilization))
record['MeasureValues'].append(prepare_measure('disk', disk_utilization))
records.append(record)
print("records {} - cpu {} - memory {} - swap {} - disk {}".format(
len(records), cpu_utilization, memory_utilization,
swap_utilization, disk_utilization))
if len(records) == 100:
write_records(records, common_attributes)
records = []
time.sleep(INTERVAL)
```
------
#### [ Node.js ]
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
async function writeRecords() {
console.log("Writing records");
const currentTime = Date.now().toString(); // Unix time in milliseconds
const dimensions = [
{'Name': 'region', 'Value': 'us-east-1'},
{'Name': 'az', 'Value': 'az1'},
{'Name': 'hostname', 'Value': 'host1'}
];
const record = {
'Dimensions': dimensions,
'MeasureName': 'metrics',
'MeasureValues': [
{
'Name': 'cpu_utilization',
'Value': '40',
'Type': 'DOUBLE',
},
{
'Name': 'memory_utilization',
'Value': '13.5',
'Type': 'DOUBLE',
},
],
'MeasureValueType': 'MULTI',
'Time': currentTime.toString()
}
const records = [record];
const params = {
DatabaseName: 'DatabaseName',
TableName: 'TableName',
Records: records
};
const response = await writeClient.writeRecords(params);
console.log(response);
}
```
------
#### [ .NET ]
```
using System;
using System.IO;
using System.Collections.Generic;
using Amazon.TimestreamWrite;
using Amazon.TimestreamWrite.Model;
using System.Threading.Tasks;
namespace TimestreamDotNetSample
{
static class MultiMeasureValueConstants
{
public const string MultiMeasureValueSampleDb = "multiMeasureValueSampleDb";
public const string MultiMeasureValueSampleTable = "multiMeasureValueSampleTable";
}
public class MultiValueAttributesExample
{
private readonly AmazonTimestreamWriteClient writeClient;
public MultiValueAttributesExample(AmazonTimestreamWriteClient writeClient)
{
this.writeClient = writeClient;
}
public async Task WriteRecordsMultiMeasureValueSingleRecord()
{
Console.WriteLine("Writing records with multi value attributes");
DateTimeOffset now = DateTimeOffset.UtcNow;
string currentTimeString = (now.ToUnixTimeMilliseconds()).ToString();
List dimensions = new List{
new Dimension { Name = "region", Value = "us-east-1" },
new Dimension { Name = "az", Value = "az1" },
new Dimension { Name = "hostname", Value = "host1" }
};
var commonAttributes = new Record
{
Dimensions = dimensions,
Time = currentTimeString
};
var cpuUtilization = new MeasureValue
{
Name = "cpu_utilization",
Value = "13.6",
Type = "DOUBLE"
};
var memoryUtilization = new MeasureValue
{
Name = "memory_utilization",
Value = "40",
Type = "DOUBLE"
};
var computationalRecord = new Record
{
MeasureName = "cpu_memory",
MeasureValues = new List {cpuUtilization, memoryUtilization},
MeasureValueType = "MULTI"
};
List records = new List();
records.Add(computationalRecord);
try
{
var writeRecordsRequest = new WriteRecordsRequest
{
DatabaseName = MultiMeasureValueConstants.MultiMeasureValueSampleDb,
TableName = MultiMeasureValueConstants.MultiMeasureValueSampleTable,
Records = records,
CommonAttributes = commonAttributes
};
WriteRecordsResponse response = await writeClient.WriteRecordsAsync(writeRecordsRequest);
Console.WriteLine($"Write records status code: {response.HttpStatusCode.ToString()}");
}
catch (Exception e)
{
Console.WriteLine("Write records failure:" + e.ToString());
}
}
public async Task WriteRecordsMultiMeasureValueMultipleRecords()
{
Console.WriteLine("Writing records with multi value attributes mixture type");
DateTimeOffset now = DateTimeOffset.UtcNow;
string currentTimeString = (now.ToUnixTimeMilliseconds()).ToString();
List dimensions = new List{
new Dimension { Name = "region", Value = "us-east-1" },
new Dimension { Name = "az", Value = "az1" },
new Dimension { Name = "hostname", Value = "host1" }
};
var commonAttributes = new Record
{
Dimensions = dimensions,
Time = currentTimeString
};
var cpuUtilization = new MeasureValue
{
Name = "cpu_utilization",
Value = "13.6",
Type = "DOUBLE"
};
var memoryUtilization = new MeasureValue
{
Name = "memory_utilization",
Value = "40",
Type = "DOUBLE"
};
var activeCores = new MeasureValue
{
Name = "active_cores",
Value = "4",
Type = "BIGINT"
};
var computationalRecord = new Record
{
MeasureName = "computational_utilization",
MeasureValues = new List {cpuUtilization, memoryUtilization, activeCores},
MeasureValueType = "MULTI"
};
var aliveRecord = new Record
{
MeasureName = "is_healthy",
MeasureValue = "true",
MeasureValueType = "BOOLEAN"
};
List records = new List();
records.Add(computationalRecord);
records.Add(aliveRecord);
try
{
var writeRecordsRequest = new WriteRecordsRequest
{
DatabaseName = MultiMeasureValueConstants.MultiMeasureValueSampleDb,
TableName = MultiMeasureValueConstants.MultiMeasureValueSampleTable,
Records = records,
CommonAttributes = commonAttributes
};
WriteRecordsResponse response = await writeClient.WriteRecordsAsync(writeRecordsRequest);
Console.WriteLine($"Write records status code: {response.HttpStatusCode.ToString()}");
}
catch (Exception e)
{
Console.WriteLine("Write records failure:" + e.ToString());
}
}
}
}
```
------
## 書き込み失敗の処理
Amazon Timestream での書き込みは、次の 1 つ以上の理由で失敗する可能性があります。
+ タイムスタンプがメモリストアの保持期間外であるレコードがある。
+ Timestream で定義された制限を超えるディメンションまたはメジャーを含むレコードがある。
+ Amazon Timestream が重複レコードを検出した。同じディメンション、タイムスタンプ、メジャー名を持つものの、以下の条件に当てはまる複数のレコードがある場合、レコードは重複としてマークされます。
+ メジャー値が異なる。
+ バージョンがリクエストに存在しないか、または新しいレコードのバージョンの値が既存の値以下である。Amazon Timestream がこの理由でデータを拒否した場合、`RejectedRecords` の `ExistingVersion` フィールドには、Amazon Timestream に保存されているレコードの最新バージョンが含まれます。強制的に更新するには、レコードのバージョンを `ExistingVersion` より大きい値に設定してリクエストを再送信します。
エラーと拒否されたレコードの詳細については、「[エラー](https://docs.aws.amazon.com/timestream/latest/developerguide/API_WriteRecords.html#API_WriteRecords_Errors)」と「[RejectedRecord](https://docs.aws.amazon.com/timestream/latest/developerguide/API_RejectedRecord.html)」を参照してください。
Timestream にレコードを書き込もうとしたときにアプリケーションが `RejectedRecordsException` を受信した場合、拒否されたレコードを解析して、次のように書き込み失敗の詳細を確認できます。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
try {
WriteRecordsResult writeRecordsResult = amazonTimestreamWrite.writeRecords(writeRecordsRequest);
System.out.println("WriteRecords Status: " + writeRecordsResult.getSdkHttpMetadata().getHttpStatusCode());
} catch (RejectedRecordsException e) {
System.out.println("RejectedRecords: " + e);
for (RejectedRecord rejectedRecord : e.getRejectedRecords()) {
System.out.println("Rejected Index " + rejectedRecord.getRecordIndex() + ": "
+ rejectedRecord.getReason());
}
System.out.println("Other records were written successfully. ");
} catch (Exception e) {
System.out.println("Error: " + e);
}
```
------
#### [ Java v2 ]
```
try {
WriteRecordsResponse writeRecordsResponse = timestreamWriteClient.writeRecords(writeRecordsRequest);
System.out.println("writeRecordsWithCommonAttributes Status: " + writeRecordsResponse.sdkHttpResponse().statusCode());
} catch (RejectedRecordsException e) {
System.out.println("RejectedRecords: " + e);
for (RejectedRecord rejectedRecord : e.rejectedRecords()) {
System.out.println("Rejected Index " + rejectedRecord.recordIndex() + ": "
+ rejectedRecord.reason());
}
System.out.println("Other records were written successfully. ");
} catch (Exception e) {
System.out.println("Error: " + e);
}
```
------
#### [ Go ]
```
_, err = writeSvc.WriteRecords(writeRecordsInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Write records is successful")
}
```
------
#### [ Python ]
```
try:
result = self.client.write_records(DatabaseName=Constant.DATABASE_NAME, TableName=Constant.TABLE_NAME, Records=records, CommonAttributes=common_attributes)
print("WriteRecords Status: [%s]" % result['ResponseMetadata']['HTTPStatusCode'])
except self.client.exceptions.RejectedRecordsException as err:
print("RejectedRecords: ", err)
for rr in err.response["RejectedRecords"]:
print("Rejected Index " + str(rr["RecordIndex"]) + ": " + rr["Reason"])
print("Other records were written successfully. ")
except Exception as err:
print("Error:", err)
```
------
#### [ Node.js ]
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
await request.promise().then(
(data) => {
console.log("Write records successful");
},
(err) => {
console.log("Error writing records:", err);
if (err.code === 'RejectedRecordsException') {
const responsePayload = JSON.parse(request.response.httpResponse.body.toString());
console.log("RejectedRecords: ", responsePayload.RejectedRecords);
console.log("Other records were written successfully. ");
}
}
);
```
------
#### [ .NET ]
```
try
{
var writeRecordsRequest = new WriteRecordsRequest
{
DatabaseName = Constants.DATABASE_NAME,
TableName = Constants.TABLE_NAME,
Records = records,
CommonAttributes = commonAttributes
};
WriteRecordsResponse response = await writeClient.WriteRecordsAsync(writeRecordsRequest);
Console.WriteLine($"Write records status code: {response.HttpStatusCode.ToString()}");
}
catch (RejectedRecordsException e) {
Console.WriteLine("RejectedRecordsException:" + e.ToString());
foreach (RejectedRecord rr in e.RejectedRecords) {
Console.WriteLine("RecordIndex " + rr.RecordIndex + " : " + rr.Reason);
}
Console.WriteLine("Other records were written successfully. ");
}
catch (Exception e)
{
Console.WriteLine("Write records failure:" + e.ToString());
}
```
------
# クエリを実行する
**Topics**
+ [結果のページ分割](#code-samples.run-query.pagination)
+ [結果セットの解析](#code-samples.run-query.parsing)
+ [クエリステータスへのアクセス](#code-samples.run-query.query-status)
## 結果のページ分割
クエリを実行すると、Timestream は結果セットをページ分割された方法で返して、アプリケーションの応答性を最適化します。以下のコードスニペットは結果セットをページ分割する方法を示しています。null 値が発生するまで、すべての結果セットページをループする必要があります。ページ分割トークンは、Timestream for LiveAnalytics によって発行されてから 3 時間後に期限切れになります。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
private void runQuery(String queryString) {
try {
QueryRequest queryRequest = new QueryRequest();
queryRequest.setQueryString(queryString);
QueryResult queryResult = queryClient.query(queryRequest);
while (true) {
parseQueryResult(queryResult);
if (queryResult.getNextToken() == null) {
break;
}
queryRequest.setNextToken(queryResult.getNextToken());
queryResult = queryClient.query(queryRequest);
}
} catch (Exception e) {
// Some queries might fail with 500 if the result of a sequence function has more than 10000 entries
e.printStackTrace();
}
}
```
------
#### [ Java v2 ]
```
private void runQuery(String queryString) {
try {
QueryRequest queryRequest = QueryRequest.builder().queryString(queryString).build();
final QueryIterable queryResponseIterator = timestreamQueryClient.queryPaginator(queryRequest);
for(QueryResponse queryResponse : queryResponseIterator) {
parseQueryResult(queryResponse);
}
} catch (Exception e) {
// Some queries might fail with 500 if the result of a sequence function has more than 10000 entries
e.printStackTrace();
}
}
```
------
#### [ Go ]
```
func runQuery(queryPtr *string, querySvc *timestreamquery.TimestreamQuery, f *os.File) {
queryInput := ×treamquery.QueryInput{
QueryString: aws.String(*queryPtr),
}
fmt.Println("QueryInput:")
fmt.Println(queryInput)
// execute the query
err := querySvc.QueryPages(queryInput,
func(page *timestreamquery.QueryOutput, lastPage bool) bool {
// process query response
queryStatus := page.QueryStatus
fmt.Println("Current query status:", queryStatus)
// query response metadata
// includes column names and types
metadata := page.ColumnInfo
// fmt.Println("Metadata:")
fmt.Println(metadata)
header := ""
for i := 0; i < len(metadata); i++ {
header += *metadata[i].Name
if i != len(metadata)-1 {
header += ", "
}
}
write(f, header)
// query response data
fmt.Println("Data:")
// process rows
rows := page.Rows
for i := 0; i < len(rows); i++ {
data := rows[i].Data
value := processRowType(data, metadata)
fmt.Println(value)
write(f, value)
}
fmt.Println("Number of rows:", len(page.Rows))
return true
})
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
}
}
```
------
#### [ Python ]
```
def run_query(self, query_string):
try:
page_iterator = self.paginator.paginate(QueryString=query_string)
for page in page_iterator:
self._parse_query_result(page)
except Exception as err:
print("Exception while running query:", err)
```
------
#### [ Node.js ]
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
async function getAllRows(query, nextToken) {
const params = {
QueryString: query
};
if (nextToken) {
params.NextToken = nextToken;
}
await queryClient.query(params).promise()
.then(
(response) => {
parseQueryResult(response);
if (response.NextToken) {
getAllRows(query, response.NextToken);
}
},
(err) => {
console.error("Error while querying:", err);
});
}
```
------
#### [ .NET ]
```
private async Task RunQueryAsync(string queryString)
{
try
{
QueryRequest queryRequest = new QueryRequest();
queryRequest.QueryString = queryString;
QueryResponse queryResponse = await queryClient.QueryAsync(queryRequest);
while (true)
{
ParseQueryResult(queryResponse);
if (queryResponse.NextToken == null)
{
break;
}
queryRequest.NextToken = queryResponse.NextToken;
queryResponse = await queryClient.QueryAsync(queryRequest);
}
} catch(Exception e)
{
// Some queries might fail with 500 if the result of a sequence function has more than 10000 entries
Console.WriteLine(e.ToString());
}
}
```
------
## 結果セットの解析
次のコードスニペットを使用して、結果セットからデータを抽出できます。クエリが完了してから 24 時間は、クエリ結果にアクセスできます。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
private static final DateTimeFormatter TIMESTAMP_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSSSSS");
private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd");
private static final DateTimeFormatter TIME_FORMATTER = DateTimeFormatter.ofPattern("HH:mm:ss.SSSSSSSSS");
private static final long ONE_GB_IN_BYTES = 1073741824L;
private void parseQueryResult(QueryResult response) {
final QueryStatus currentStatusOfQuery = queryResult.getQueryStatus();
System.out.println("Query progress so far: " + currentStatusOfQuery.getProgressPercentage() + "%");
double bytesScannedSoFar = ((double) currentStatusOfQuery.getCumulativeBytesScanned() / ONE_GB_IN_BYTES);
System.out.println("Bytes scanned so far: " + bytesScannedSoFar + " GB");
double bytesMeteredSoFar = ((double) currentStatusOfQuery.getCumulativeBytesMetered() / ONE_GB_IN_BYTES);
System.out.println("Bytes metered so far: " + bytesMeteredSoFar + " GB");
List columnInfo = response.getColumnInfo();
List rows = response.getRows();
System.out.println("Metadata: " + columnInfo);
System.out.println("Data: ");
// iterate every row
for (Row row : rows) {
System.out.println(parseRow(columnInfo, row));
}
}
private String parseRow(List columnInfo, Row row) {
List data = row.getData();
List rowOutput = new ArrayList<>();
// iterate every column per row
for (int j = 0; j < data.size(); j++) {
ColumnInfo info = columnInfo.get(j);
Datum datum = data.get(j);
rowOutput.add(parseDatum(info, datum));
}
return String.format("{%s}", rowOutput.stream().map(Object::toString).collect(Collectors.joining(",")));
}
private String parseDatum(ColumnInfo info, Datum datum) {
if (datum.isNullValue() != null && datum.isNullValue()) {
return info.getName() + "=" + "NULL";
}
Type columnType = info.getType();
// If the column is of TimeSeries Type
if (columnType.getTimeSeriesMeasureValueColumnInfo() != null) {
return parseTimeSeries(info, datum);
}
// If the column is of Array Type
else if (columnType.getArrayColumnInfo() != null) {
List arrayValues = datum.getArrayValue();
return info.getName() + "=" + parseArray(info.getType().getArrayColumnInfo(), arrayValues);
}
// If the column is of Row Type
else if (columnType.getRowColumnInfo() != null) {
List rowColumnInfo = info.getType().getRowColumnInfo();
Row rowValues = datum.getRowValue();
return parseRow(rowColumnInfo, rowValues);
}
// If the column is of Scalar Type
else {
return parseScalarType(info, datum);
}
}
private String parseTimeSeries(ColumnInfo info, Datum datum) {
List timeSeriesOutput = new ArrayList<>();
for (TimeSeriesDataPoint dataPoint : datum.getTimeSeriesValue()) {
timeSeriesOutput.add("{time=" + dataPoint.getTime() + ", value=" +
parseDatum(info.getType().getTimeSeriesMeasureValueColumnInfo(), dataPoint.getValue()) + "}");
}
return String.format("[%s]", timeSeriesOutput.stream().map(Object::toString).collect(Collectors.joining(",")));
}
private String parseScalarType(ColumnInfo info, Datum datum) {
switch (ScalarType.fromValue(info.getType().getScalarType())) {
case VARCHAR:
return parseColumnName(info) + datum.getScalarValue();
case BIGINT:
Long longValue = Long.valueOf(datum.getScalarValue());
return parseColumnName(info) + longValue;
case INTEGER:
Integer intValue = Integer.valueOf(datum.getScalarValue());
return parseColumnName(info) + intValue;
case BOOLEAN:
Boolean booleanValue = Boolean.valueOf(datum.getScalarValue());
return parseColumnName(info) + booleanValue;
case DOUBLE:
Double doubleValue = Double.valueOf(datum.getScalarValue());
return parseColumnName(info) + doubleValue;
case TIMESTAMP:
return parseColumnName(info) + LocalDateTime.parse(datum.getScalarValue(), TIMESTAMP_FORMATTER);
case DATE:
return parseColumnName(info) + LocalDate.parse(datum.getScalarValue(), DATE_FORMATTER);
case TIME:
return parseColumnName(info) + LocalTime.parse(datum.getScalarValue(), TIME_FORMATTER);
case INTERVAL_DAY_TO_SECOND:
case INTERVAL_YEAR_TO_MONTH:
return parseColumnName(info) + datum.getScalarValue();
case UNKNOWN:
return parseColumnName(info) + datum.getScalarValue();
default:
throw new IllegalArgumentException("Given type is not valid: " + info.getType().getScalarType());
}
}
private String parseColumnName(ColumnInfo info) {
return info.getName() == null ? "" : info.getName() + "=";
}
private String parseArray(ColumnInfo arrayColumnInfo, List arrayValues) {
List arrayOutput = new ArrayList<>();
for (Datum datum : arrayValues) {
arrayOutput.add(parseDatum(arrayColumnInfo, datum));
}
return String.format("[%s]", arrayOutput.stream().map(Object::toString).collect(Collectors.joining(",")));
}
```
------
#### [ Java v2 ]
```
private static final long ONE_GB_IN_BYTES = 1073741824L;
private void parseQueryResult(QueryResponse response) {
final QueryStatus currentStatusOfQuery = response.queryStatus();
System.out.println("Query progress so far: " + currentStatusOfQuery.progressPercentage() + "%");
double bytesScannedSoFar = ((double) currentStatusOfQuery.cumulativeBytesScanned() / ONE_GB_IN_BYTES);
System.out.println("Bytes scanned so far: " + bytesScannedSoFar + " GB");
double bytesMeteredSoFar = ((double) currentStatusOfQuery.cumulativeBytesMetered() / ONE_GB_IN_BYTES);
System.out.println("Bytes metered so far: " + bytesMeteredSoFar + " GB");
List columnInfo = response.columnInfo();
List rows = response.rows();
System.out.println("Metadata: " + columnInfo);
System.out.println("Data: ");
// iterate every row
for (Row row : rows) {
System.out.println(parseRow(columnInfo, row));
}
}
private String parseRow(List columnInfo, Row row) {
List data = row.data();
List rowOutput = new ArrayList<>();
// iterate every column per row
for (int j = 0; j < data.size(); j++) {
ColumnInfo info = columnInfo.get(j);
Datum datum = data.get(j);
rowOutput.add(parseDatum(info, datum));
}
return String.format("{%s}", rowOutput.stream().map(Object::toString).collect(Collectors.joining(",")));
}
private String parseDatum(ColumnInfo info, Datum datum) {
if (datum.nullValue() != null && datum.nullValue()) {
return info.name() + "=" + "NULL";
}
Type columnType = info.type();
// If the column is of TimeSeries Type
if (columnType.timeSeriesMeasureValueColumnInfo() != null) {
return parseTimeSeries(info, datum);
}
// If the column is of Array Type
else if (columnType.arrayColumnInfo() != null) {
List arrayValues = datum.arrayValue();
return info.name() + "=" + parseArray(info.type().arrayColumnInfo(), arrayValues);
}
// If the column is of Row Type
else if (columnType.rowColumnInfo() != null && columnType.rowColumnInfo().size() > 0) {
List rowColumnInfo = info.type().rowColumnInfo();
Row rowValues = datum.rowValue();
return parseRow(rowColumnInfo, rowValues);
}
// If the column is of Scalar Type
else {
return parseScalarType(info, datum);
}
}
private String parseTimeSeries(ColumnInfo info, Datum datum) {
List timeSeriesOutput = new ArrayList<>();
for (TimeSeriesDataPoint dataPoint : datum.timeSeriesValue()) {
timeSeriesOutput.add("{time=" + dataPoint.time() + ", value=" +
parseDatum(info.type().timeSeriesMeasureValueColumnInfo(), dataPoint.value()) + "}");
}
return String.format("[%s]", timeSeriesOutput.stream().map(Object::toString).collect(Collectors.joining(",")));
}
private String parseScalarType(ColumnInfo info, Datum datum) {
return parseColumnName(info) + datum.scalarValue();
}
private String parseColumnName(ColumnInfo info) {
return info.name() == null ? "" : info.name() + "=";
}
private String parseArray(ColumnInfo arrayColumnInfo, List arrayValues) {
List arrayOutput = new ArrayList<>();
for (Datum datum : arrayValues) {
arrayOutput.add(parseDatum(arrayColumnInfo, datum));
}
return String.format("[%s]", arrayOutput.stream().map(Object::toString).collect(Collectors.joining(",")));
}
```
------
#### [ Go ]
```
func processScalarType(data *timestreamquery.Datum) string {
return *data.ScalarValue
}
func processTimeSeriesType(data []*timestreamquery.TimeSeriesDataPoint, columnInfo *timestreamquery.ColumnInfo) string {
value := ""
for k := 0; k < len(data); k++ {
time := data[k].Time
value += *time + ":"
if columnInfo.Type.ScalarType != nil {
value += processScalarType(data[k].Value)
} else if columnInfo.Type.ArrayColumnInfo != nil {
value += processArrayType(data[k].Value.ArrayValue, columnInfo.Type.ArrayColumnInfo)
} else if columnInfo.Type.RowColumnInfo != nil {
value += processRowType(data[k].Value.RowValue.Data, columnInfo.Type.RowColumnInfo)
} else {
fail("Bad data type")
}
if k != len(data)-1 {
value += ", "
}
}
return value
}
func processArrayType(datumList []*timestreamquery.Datum, columnInfo *timestreamquery.ColumnInfo) string {
value := ""
for k := 0; k < len(datumList); k++ {
if columnInfo.Type.ScalarType != nil {
value += processScalarType(datumList[k])
} else if columnInfo.Type.TimeSeriesMeasureValueColumnInfo != nil {
value += processTimeSeriesType(datumList[k].TimeSeriesValue, columnInfo.Type.TimeSeriesMeasureValueColumnInfo)
} else if columnInfo.Type.ArrayColumnInfo != nil {
value += "["
value += processArrayType(datumList[k].ArrayValue, columnInfo.Type.ArrayColumnInfo)
value += "]"
} else if columnInfo.Type.RowColumnInfo != nil {
value += "["
value += processRowType(datumList[k].RowValue.Data, columnInfo.Type.RowColumnInfo)
value += "]"
} else {
fail("Bad column type")
}
if k != len(datumList)-1 {
value += ", "
}
}
return value
}
func processRowType(data []*timestreamquery.Datum, metadata []*timestreamquery.ColumnInfo) string {
value := ""
for j := 0; j < len(data); j++ {
if metadata[j].Type.ScalarType != nil {
// process simple data types
value += processScalarType(data[j])
} else if metadata[j].Type.TimeSeriesMeasureValueColumnInfo != nil {
// fmt.Println("Timeseries measure value column info")
// fmt.Println(metadata[j].Type.TimeSeriesMeasureValueColumnInfo.Type)
datapointList := data[j].TimeSeriesValue
value += "["
value += processTimeSeriesType(datapointList, metadata[j].Type.TimeSeriesMeasureValueColumnInfo)
value += "]"
} else if metadata[j].Type.ArrayColumnInfo != nil {
columnInfo := metadata[j].Type.ArrayColumnInfo
// fmt.Println("Array column info")
// fmt.Println(columnInfo)
datumList := data[j].ArrayValue
value += "["
value += processArrayType(datumList, columnInfo)
value += "]"
} else if metadata[j].Type.RowColumnInfo != nil {
columnInfo := metadata[j].Type.RowColumnInfo
datumList := data[j].RowValue.Data
value += "["
value += processRowType(datumList, columnInfo)
value += "]"
} else {
panic("Bad column type")
}
// comma seperated column values
if j != len(data)-1 {
value += ", "
}
}
return value
}
```
------
#### [ Python ]
```
def _parse_query_result(self, query_result):
query_status = query_result["QueryStatus"]
progress_percentage = query_status["ProgressPercentage"]
print(f"Query progress so far: {progress_percentage}%")
bytes_scanned = float(query_status["CumulativeBytesScanned"]) / ONE_GB_IN_BYTES
print(f"Data scanned so far: {bytes_scanned} GB")
bytes_metered = float(query_status["CumulativeBytesMetered"]) / ONE_GB_IN_BYTES
print(f"Data metered so far: {bytes_metered} GB")
column_info = query_result['ColumnInfo']
print("Metadata: %s" % column_info)
print("Data: ")
for row in query_result['Rows']:
print(self._parse_row(column_info, row))
def _parse_row(self, column_info, row):
data = row['Data']
row_output = []
for j in range(len(data)):
info = column_info[j]
datum = data[j]
row_output.append(self._parse_datum(info, datum))
return "{%s}" % str(row_output)
def _parse_datum(self, info, datum):
if datum.get('NullValue', False):
return "%s=NULL" % info['Name'],
column_type = info['Type']
# If the column is of TimeSeries Type
if 'TimeSeriesMeasureValueColumnInfo' in column_type:
return self._parse_time_series(info, datum)
# If the column is of Array Type
elif 'ArrayColumnInfo' in column_type:
array_values = datum['ArrayValue']
return "%s=%s" % (info['Name'], self._parse_array(info['Type']['ArrayColumnInfo'], array_values))
# If the column is of Row Type
elif 'RowColumnInfo' in column_type:
row_column_info = info['Type']['RowColumnInfo']
row_values = datum['RowValue']
return self._parse_row(row_column_info, row_values)
# If the column is of Scalar Type
else:
return self._parse_column_name(info) + datum['ScalarValue']
def _parse_time_series(self, info, datum):
time_series_output = []
for data_point in datum['TimeSeriesValue']:
time_series_output.append("{time=%s, value=%s}"
% (data_point['Time'],
self._parse_datum(info['Type']['TimeSeriesMeasureValueColumnInfo'],
data_point['Value'])))
return "[%s]" % str(time_series_output)
def _parse_array(self, array_column_info, array_values):
array_output = []
for datum in array_values:
array_output.append(self._parse_datum(array_column_info, datum))
return "[%s]" % str(array_output)
@staticmethod
def _parse_column_name(info):
if 'Name' in info:
return info['Name'] + "="
else:
return ""
```
------
#### [ Node.js ]
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
function parseQueryResult(response) {
const queryStatus = response.QueryStatus;
console.log("Current query status: " + JSON.stringify(queryStatus));
const columnInfo = response.ColumnInfo;
const rows = response.Rows;
console.log("Metadata: " + JSON.stringify(columnInfo));
console.log("Data: ");
rows.forEach(function (row) {
console.log(parseRow(columnInfo, row));
});
}
function parseRow(columnInfo, row) {
const data = row.Data;
const rowOutput = [];
var i;
for ( i = 0; i < data.length; i++ ) {
info = columnInfo[i];
datum = data[i];
rowOutput.push(parseDatum(info, datum));
}
return `{${rowOutput.join(", ")}}`
}
function parseDatum(info, datum) {
if (datum.NullValue != null && datum.NullValue === true) {
return `${info.Name}=NULL`;
}
const columnType = info.Type;
// If the column is of TimeSeries Type
if (columnType.TimeSeriesMeasureValueColumnInfo != null) {
return parseTimeSeries(info, datum);
}
// If the column is of Array Type
else if (columnType.ArrayColumnInfo != null) {
const arrayValues = datum.ArrayValue;
return `${info.Name}=${parseArray(info.Type.ArrayColumnInfo, arrayValues)}`;
}
// If the column is of Row Type
else if (columnType.RowColumnInfo != null) {
const rowColumnInfo = info.Type.RowColumnInfo;
const rowValues = datum.RowValue;
return parseRow(rowColumnInfo, rowValues);
}
// If the column is of Scalar Type
else {
return parseScalarType(info, datum);
}
}
function parseTimeSeries(info, datum) {
const timeSeriesOutput = [];
datum.TimeSeriesValue.forEach(function (dataPoint) {
timeSeriesOutput.push(`{time=${dataPoint.Time}, value=${parseDatum(info.Type.TimeSeriesMeasureValueColumnInfo, dataPoint.Value)}}`)
});
return `[${timeSeriesOutput.join(", ")}]`
}
function parseScalarType(info, datum) {
return parseColumnName(info) + datum.ScalarValue;
}
function parseColumnName(info) {
return info.Name == null ? "" : `${info.Name}=`;
}
function parseArray(arrayColumnInfo, arrayValues) {
const arrayOutput = [];
arrayValues.forEach(function (datum) {
arrayOutput.push(parseDatum(arrayColumnInfo, datum));
});
return `[${arrayOutput.join(", ")}]`
}
```
------
#### [ .NET ]
```
private void ParseQueryResult(QueryResponse response)
{
List columnInfo = response.ColumnInfo;
var options = new JsonSerializerOptions
{
IgnoreNullValues = true
};
List columnInfoStrings = columnInfo.ConvertAll(x => JsonSerializer.Serialize(x, options));
List rows = response.Rows;
QueryStatus queryStatus = response.QueryStatus;
Console.WriteLine("Current Query status:" + JsonSerializer.Serialize(queryStatus, options));
Console.WriteLine("Metadata:" + string.Join(",", columnInfoStrings));
Console.WriteLine("Data:");
foreach (Row row in rows)
{
Console.WriteLine(ParseRow(columnInfo, row));
}
}
private string ParseRow(List columnInfo, Row row)
{
List data = row.Data;
List rowOutput = new List();
for (int j = 0; j < data.Count; j++)
{
ColumnInfo info = columnInfo[j];
Datum datum = data[j];
rowOutput.Add(ParseDatum(info, datum));
}
return $"{{{string.Join(",", rowOutput)}}}";
}
private string ParseDatum(ColumnInfo info, Datum datum)
{
if (datum.NullValue)
{
return $"{info.Name}=NULL";
}
Amazon.TimestreamQuery.Model.Type columnType = info.Type;
if (columnType.TimeSeriesMeasureValueColumnInfo != null)
{
return ParseTimeSeries(info, datum);
}
else if (columnType.ArrayColumnInfo != null)
{
List arrayValues = datum.ArrayValue;
return $"{info.Name}={ParseArray(info.Type.ArrayColumnInfo, arrayValues)}";
}
else if (columnType.RowColumnInfo != null && columnType.RowColumnInfo.Count > 0)
{
List rowColumnInfo = info.Type.RowColumnInfo;
Row rowValue = datum.RowValue;
return ParseRow(rowColumnInfo, rowValue);
}
else
{
return ParseScalarType(info, datum);
}
}
private string ParseTimeSeries(ColumnInfo info, Datum datum)
{
var timeseriesString = datum.TimeSeriesValue
.Select(value => $"{{time={value.Time}, value={ParseDatum(info.Type.TimeSeriesMeasureValueColumnInfo, value.Value)}}}")
.Aggregate((current, next) => current + "," + next);
return $"[{timeseriesString}]";
}
private string ParseScalarType(ColumnInfo info, Datum datum)
{
return ParseColumnName(info) + datum.ScalarValue;
}
private string ParseColumnName(ColumnInfo info)
{
return info.Name == null ? "" : (info.Name + "=");
}
private string ParseArray(ColumnInfo arrayColumnInfo, List arrayValues)
{
return $"[{arrayValues.Select(value => ParseDatum(arrayColumnInfo, value)).Aggregate((current, next) => current + "," + next)}]";
}
```
------
## クエリステータスへのアクセス
クエリのステータスには `QueryResponse` からアクセスできます。クエリの進行状況、クエリによってスキャンされたバイト、クエリによって計測されたバイトに関する情報を含みます。`bytesMetered` と `bytesScanned` 値は累積的であり、クエリ結果のページ分割中に継続的に更新されます。この情報を使用して、個々のクエリによってスキャンされたバイトを理解し、それを使用して特定の決定を行うことができます。例えば、クエリの料金がスキャンされた GB あたり 0.01 USD であるとすると、クエリあたり 25 USD を超えるクエリ、または `X` GB をキャンセルできます。以下のコードスニペットは、これを行う方法を示しています。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
private static final long ONE_GB_IN_BYTES = 1073741824L;
private static final double QUERY_COST_PER_GB_IN_DOLLARS = 0.01; // Assuming the price of query is $0.01 per GB
public void cancelQueryBasedOnQueryStatus() {
System.out.println("Starting query: " + SELECT_ALL_QUERY);
QueryRequest queryRequest = new QueryRequest();
queryRequest.setQueryString(SELECT_ALL_QUERY);
QueryResult queryResult = queryClient.query(queryRequest);
while (true) {
final QueryStatus currentStatusOfQuery = queryResult.getQueryStatus();
System.out.println("Query progress so far: " + currentStatusOfQuery.getProgressPercentage() + "%");
double bytesMeteredSoFar = ((double) currentStatusOfQuery.getCumulativeBytesMetered() / ONE_GB_IN_BYTES);
System.out.println("Bytes metered so far: " + bytesMeteredSoFar + " GB");
// Cancel query if its costing more than 1 cent
if (bytesMeteredSoFar * QUERY_COST_PER_GB_IN_DOLLARS > 0.01) {
cancelQuery(queryResult);
break;
}
if (queryResult.getNextToken() == null) {
break;
}
queryRequest.setNextToken(queryResult.getNextToken());
queryResult = queryClient.query(queryRequest);
}
}
```
------
#### [ Java v2 ]
```
private static final long ONE_GB_IN_BYTES = 1073741824L;
private static final double QUERY_COST_PER_GB_IN_DOLLARS = 0.01; // Assuming the price of query is $0.01 per GB
public void cancelQueryBasedOnQueryStatus() {
System.out.println("Starting query: " + SELECT_ALL_QUERY);
QueryRequest queryRequest = QueryRequest.builder().queryString(SELECT_ALL_QUERY).build();
final QueryIterable queryResponseIterator = timestreamQueryClient.queryPaginator(queryRequest);
for(QueryResponse queryResponse : queryResponseIterator) {
final QueryStatus currentStatusOfQuery = queryResponse.queryStatus();
System.out.println("Query progress so far: " + currentStatusOfQuery.progressPercentage() + "%");
double bytesMeteredSoFar = ((double) currentStatusOfQuery.cumulativeBytesMetered() / ONE_GB_IN_BYTES);
System.out.println("Bytes metered so far: " + bytesMeteredSoFar + "GB");
// Cancel query if its costing more than 1 cent
if (bytesMeteredSoFar * QUERY_COST_PER_GB_IN_DOLLARS > 0.01) {
cancelQuery(queryResponse);
break;
}
}
}
```
------
#### [ Go ]
```
const OneGbInBytes = 1073741824
// Assuming the price of query is $0.01 per GB
const QueryCostPerGbInDollars = 0.01
func cancelQueryBasedOnQueryStatus(queryPtr *string, querySvc *timestreamquery.TimestreamQuery, f *os.File) {
queryInput := ×treamquery.QueryInput{
QueryString: aws.String(*queryPtr),
}
fmt.Println("QueryInput:")
fmt.Println(queryInput)
// execute the query
err := querySvc.QueryPages(queryInput,
func(page *timestreamquery.QueryOutput, lastPage bool) bool {
// process query response
queryStatus := page.QueryStatus
fmt.Println("Current query status:", queryStatus)
bytes_metered := float64(*queryStatus.CumulativeBytesMetered) / float64(ONE_GB_IN_BYTES)
if bytes_metered * QUERY_COST_PER_GB_IN_DOLLARS > 0.01 {
cancelQuery(page, querySvc)
return true
}
// query response metadata
// includes column names and types
metadata := page.ColumnInfo
// fmt.Println("Metadata:")
fmt.Println(metadata)
header := ""
for i := 0; i < len(metadata); i++ {
header += *metadata[i].Name
if i != len(metadata)-1 {
header += ", "
}
}
write(f, header)
// query response data
fmt.Println("Data:")
// process rows
rows := page.Rows
for i := 0; i < len(rows); i++ {
data := rows[i].Data
value := processRowType(data, metadata)
fmt.Println(value)
write(f, value)
}
fmt.Println("Number of rows:", len(page.Rows))
return true
})
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
}
}
```
------
#### [ Python ]
```
ONE_GB_IN_BYTES = 1073741824
# Assuming the price of query is $0.01 per GB
QUERY_COST_PER_GB_IN_DOLLARS = 0.01
def cancel_query_based_on_query_status(self):
try:
print("Starting query: " + self.SELECT_ALL)
page_iterator = self.paginator.paginate(QueryString=self.SELECT_ALL)
for page in page_iterator:
query_status = page["QueryStatus"]
progress_percentage = query_status["ProgressPercentage"]
print("Query progress so far: " + str(progress_percentage) + "%")
bytes_metered = query_status["CumulativeBytesMetered"] / self.ONE_GB_IN_BYTES
print("Bytes Metered so far: " + str(bytes_metered) + " GB")
if bytes_metered * self.QUERY_COST_PER_GB_IN_DOLLARS > 0.01:
self.cancel_query_for(page)
break
except Exception as err:
print("Exception while running query:", err)
traceback.print_exc(file=sys.stderr)
```
------
#### [ Node.js ]
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
function parseQueryResult(response) {
const queryStatus = response.QueryStatus;
console.log("Current query status: " + JSON.stringify(queryStatus));
const columnInfo = response.ColumnInfo;
const rows = response.Rows;
console.log("Metadata: " + JSON.stringify(columnInfo));
console.log("Data: ");
rows.forEach(function (row) {
console.log(parseRow(columnInfo, row));
});
}
function parseRow(columnInfo, row) {
const data = row.Data;
const rowOutput = [];
var i;
for ( i = 0; i < data.length; i++ ) {
info = columnInfo[i];
datum = data[i];
rowOutput.push(parseDatum(info, datum));
}
return `{${rowOutput.join(", ")}}`
}
function parseDatum(info, datum) {
if (datum.NullValue != null && datum.NullValue === true) {
return `${info.Name}=NULL`;
}
const columnType = info.Type;
// If the column is of TimeSeries Type
if (columnType.TimeSeriesMeasureValueColumnInfo != null) {
return parseTimeSeries(info, datum);
}
// If the column is of Array Type
else if (columnType.ArrayColumnInfo != null) {
const arrayValues = datum.ArrayValue;
return `${info.Name}=${parseArray(info.Type.ArrayColumnInfo, arrayValues)}`;
}
// If the column is of Row Type
else if (columnType.RowColumnInfo != null) {
const rowColumnInfo = info.Type.RowColumnInfo;
const rowValues = datum.RowValue;
return parseRow(rowColumnInfo, rowValues);
}
// If the column is of Scalar Type
else {
return parseScalarType(info, datum);
}
}
function parseTimeSeries(info, datum) {
const timeSeriesOutput = [];
datum.TimeSeriesValue.forEach(function (dataPoint) {
timeSeriesOutput.push(`{time=${dataPoint.Time}, value=${parseDatum(info.Type.TimeSeriesMeasureValueColumnInfo, dataPoint.Value)}}`)
});
return `[${timeSeriesOutput.join(", ")}]`
}
function parseScalarType(info, datum) {
return parseColumnName(info) + datum.ScalarValue;
}
function parseColumnName(info) {
return info.Name == null ? "" : `${info.Name}=`;
}
function parseArray(arrayColumnInfo, arrayValues) {
const arrayOutput = [];
arrayValues.forEach(function (datum) {
arrayOutput.push(parseDatum(arrayColumnInfo, datum));
});
return `[${arrayOutput.join(", ")}]`
}
```
------
#### [ .NET ]
```
private static readonly long ONE_GB_IN_BYTES = 1073741824L;
private static readonly double QUERY_COST_PER_GB_IN_DOLLARS = 0.01; // Assuming the price of query is $0.01 per GB
private async Task CancelQueryBasedOnQueryStatus(string queryString)
{
try
{
QueryRequest queryRequest = new QueryRequest();
queryRequest.QueryString = queryString;
QueryResponse queryResponse = await queryClient.QueryAsync(queryRequest);
while (true)
{
QueryStatus queryStatus = queryResponse.QueryStatus;
double bytesMeteredSoFar = ((double) queryStatus.CumulativeBytesMetered / ONE_GB_IN_BYTES);
// Cancel query if its costing more than 1 cent
if (bytesMeteredSoFar * QUERY_COST_PER_GB_IN_DOLLARS > 0.01)
{
await CancelQuery(queryResponse);
break;
}
ParseQueryResult(queryResponse);
if (queryResponse.NextToken == null)
{
break;
}
queryRequest.NextToken = queryResponse.NextToken;
queryResponse = await queryClient.QueryAsync(queryRequest);
}
} catch(Exception e)
{
// Some queries might fail with 500 if the result of a sequence function has more than 10000 entries
Console.WriteLine(e.ToString());
}
}
```
------
クエリをキャンセルする方法の詳細については、「[クエリをキャンセルする](code-samples.cancel-query.md)」を参照してください。
# UNLOAD クエリを実行する
次のコード例では、UNLOAD クエリを呼び出します。`UNLOAD` の詳細については「[UNLOAD を使用してクエリ結果を Timestream for LiveAnalytics から S3 にエクスポートする](export-unload.md)」を参照してください。`UNLOAD` クエリの例については、「[Timestream for LiveAnalytics の UNLOAD のユースケース例](export-unload-example-use-case.md)」を参照してください。
**Topics**
+ [UNLOAD クエリを構築して実行する](#code-samples.run-query-unload-build-and-run)
+ [UNLOAD レスポンスを解析し、行数、マニフェストリンク、メタデータリンクを取得する](#code-samples.run-query-unload-parse-response)
+ [マニフェストコンテンツの読み取りと解析](#code-samples.run-query-unload-parse-manifest)
+ [メタデータコンテンツの読み取りと解析](#code-samples.run-query-unload-parse-metadata)
## UNLOAD クエリを構築して実行する
------
#### [ Java ]
```
// When you have a SELECT like below
String QUERY_1 = "SELECT user_id, ip_address, event, session_id, measure_name, time, query, quantity, product_id, channel FROM "
+ DATABASE_NAME + "." + UNLOAD_TABLE_NAME
+ " WHERE time BETWEEN ago(2d) AND now()";
// You can construct UNLOAD query as follows
UnloadQuery unloadQuery = UnloadQuery.builder()
.selectQuery(QUERY_1)
.bucketName("timestream-sample--")
.resultsPrefix("without_partition")
.format(CSV)
.compression(UnloadQuery.Compression.GZIP)
.build();
QueryResult unloadResult = runQuery(unloadQuery.getUnloadQuery());
// Run UNLOAD query (Similar to how you run SELECT query)
// https://docs.aws.amazon.com/timestream/latest/developerguide/code-samples.run-query.html#code-samples.run-query.pagination
private QueryResult runQuery(String queryString) {
QueryResult queryResult = null;
try {
QueryRequest queryRequest = new QueryRequest();
queryRequest.setQueryString(queryString);
queryResult = queryClient.query(queryRequest);
while (true) {
parseQueryResult(queryResult);
if (queryResult.getNextToken() == null) {
break;
}
queryRequest.setNextToken(queryResult.getNextToken());
queryResult = queryClient.query(queryRequest);
}
} catch (Exception e) {
// Some queries might fail with 500 if the result of a sequence function has more than 10000 entries
e.printStackTrace();
}
return queryResult;
}
// Utility that helps to construct UNLOAD query
@Builder
static class UnloadQuery {
private String selectQuery;
private String bucketName;
private String resultsPrefix;
private Format format;
private Compression compression;
private EncryptionType encryptionType;
private List partitionColumns;
private String kmsKey;
private Character csvFieldDelimiter;
private Character csvEscapeCharacter;
public String getUnloadQuery() {
String destination = constructDestination();
String withClause = constructOptionalParameters();
return String.format("UNLOAD (%s) TO '%s' %s", selectQuery, destination, withClause);
}
private String constructDestination() {
return "s3://" + this.bucketName + "/" + this.resultsPrefix + "/";
}
private String constructOptionalParameters() {
boolean isOptionalParametersPresent = Objects.nonNull(format)
|| Objects.nonNull(compression)
|| Objects.nonNull(encryptionType)
|| Objects.nonNull(partitionColumns)
|| Objects.nonNull(kmsKey)
|| Objects.nonNull(csvFieldDelimiter)
|| Objects.nonNull(csvEscapeCharacter);
String withClause = "";
if (isOptionalParametersPresent) {
StringJoiner optionalParameters = new StringJoiner(",");
if (Objects.nonNull(format)) {
optionalParameters.add("format = '" + format + "'");
}
if (Objects.nonNull(compression)) {
optionalParameters.add("compression = '" + compression + "'");
}
if (Objects.nonNull(encryptionType)) {
optionalParameters.add("encryption = '" + encryptionType + "'");
}
if (Objects.nonNull(kmsKey)) {
optionalParameters.add("kms_key = '" + kmsKey + "'");
}
if (Objects.nonNull(csvFieldDelimiter)) {
optionalParameters.add("field_delimiter = '" + csvFieldDelimiter + "'");
}
if (Objects.nonNull(csvEscapeCharacter)) {
optionalParameters.add("escaped_by = '" + csvEscapeCharacter + "'");
}
if (Objects.nonNull(partitionColumns) && !partitionColumns.isEmpty()) {
final StringJoiner partitionedByList = new StringJoiner(",");
partitionColumns.forEach(column -> partitionedByList.add("'" + column + "'"));
optionalParameters.add(String.format("partitioned_by = ARRAY[%s]", partitionedByList));
}
withClause = String.format("WITH (%s)", optionalParameters);
}
return withClause;
}
public enum Format {
CSV, PARQUET
}
public enum Compression {
GZIP, NONE
}
public enum EncryptionType {
SSE_S3, SSE_KMS
}
@Override
public String toString() {
return getUnloadQuery();
}
}
```
------
#### [ Java v2 ]
```
// When you have a SELECT like below
String QUERY_1 = "SELECT user_id, ip_address, event, session_id, measure_name, time, query, quantity, product_id, channel FROM "
+ DATABASE_NAME + "." + UNLOAD_TABLE_NAME
+ " WHERE time BETWEEN ago(2d) AND now()";
//You can construct UNLOAD query as follows
UnloadQuery unloadQuery = UnloadQuery.builder()
.selectQuery(QUERY_1)
.bucketName("timestream-sample--")
.resultsPrefix("without_partition")
.format(CSV)
.compression(UnloadQuery.Compression.GZIP)
.build();
QueryResponse unloadResponse = runQuery(unloadQuery.getUnloadQuery());
// Run UNLOAD query (Similar to how you run SELECT query)
// https://docs.aws.amazon.com/timestream/latest/developerguide/code-samples.run-query.html#code-samples.run-query.pagination
private QueryResponse runQuery(String queryString) {
QueryResponse finalResponse = null;
try {
QueryRequest queryRequest = QueryRequest.builder().queryString(queryString).build();
final QueryIterable queryResponseIterator = timestreamQueryClient.queryPaginator(queryRequest);
for(QueryResponse queryResponse : queryResponseIterator) {
parseQueryResult(queryResponse);
finalResponse = queryResponse;
}
} catch (Exception e) {
// Some queries might fail with 500 if the result of a sequence function has more than 10000 entries
e.printStackTrace();
}
return finalResponse;
}
// Utility that helps to construct UNLOAD query
@Builder
static class UnloadQuery {
private String selectQuery;
private String bucketName;
private String resultsPrefix;
private Format format;
private Compression compression;
private EncryptionType encryptionType;
private List partitionColumns;
private String kmsKey;
private Character csvFieldDelimiter;
private Character csvEscapeCharacter;
public String getUnloadQuery() {
String destination = constructDestination();
String withClause = constructOptionalParameters();
return String.format("UNLOAD (%s) TO '%s' %s", selectQuery, destination, withClause);
}
private String constructDestination() {
return "s3://" + this.bucketName + "/" + this.resultsPrefix + "/";
}
private String constructOptionalParameters() {
boolean isOptionalParametersPresent = Objects.nonNull(format)
|| Objects.nonNull(compression)
|| Objects.nonNull(encryptionType)
|| Objects.nonNull(partitionColumns)
|| Objects.nonNull(kmsKey)
|| Objects.nonNull(csvFieldDelimiter)
|| Objects.nonNull(csvEscapeCharacter);
String withClause = "";
if (isOptionalParametersPresent) {
StringJoiner optionalParameters = new StringJoiner(",");
if (Objects.nonNull(format)) {
optionalParameters.add("format = '" + format + "'");
}
if (Objects.nonNull(compression)) {
optionalParameters.add("compression = '" + compression + "'");
}
if (Objects.nonNull(encryptionType)) {
optionalParameters.add("encryption = '" + encryptionType + "'");
}
if (Objects.nonNull(kmsKey)) {
optionalParameters.add("kms_key = '" + kmsKey + "'");
}
if (Objects.nonNull(csvFieldDelimiter)) {
optionalParameters.add("field_delimiter = '" + csvFieldDelimiter + "'");
}
if (Objects.nonNull(csvEscapeCharacter)) {
optionalParameters.add("escaped_by = '" + csvEscapeCharacter + "'");
}
if (Objects.nonNull(partitionColumns) && !partitionColumns.isEmpty()) {
final StringJoiner partitionedByList = new StringJoiner(",");
partitionColumns.forEach(column -> partitionedByList.add("'" + column + "'"));
optionalParameters.add(String.format("partitioned_by = ARRAY[%s]", partitionedByList));
}
withClause = String.format("WITH (%s)", optionalParameters);
}
return withClause;
}
public enum Format {
CSV, PARQUET
}
public enum Compression {
GZIP, NONE
}
public enum EncryptionType {
SSE_S3, SSE_KMS
}
@Override
public String toString() {
return getUnloadQuery();
}
}
```
------
#### [ Go ]
```
// When you have a SELECT like below
var Query = "SELECT user_id, ip_address, event, session_id, measure_name, time, query, quantity, product_id, channel FROM "
+ *databaseName + "." + *tableName + " WHERE time BETWEEN ago(2d) AND now()"
// You can construct UNLOAD query as follows
var unloadQuery = UnloadQuery{
Query: "SELECT user_id, ip_address, session_id, measure_name, time, query, quantity, product_id, channel, event FROM " + *databaseName + "." + *tableName +
" WHERE time BETWEEN ago(2d) AND now()",
Partitioned_by: []string{},
Compression: "GZIP",
Format: "CSV",
S3Location: bucketName,
ResultPrefix: "without_partition",
}
// Run UNLOAD query (Similar to how you run SELECT query)
// https://docs.aws.amazon.com/timestream/latest/developerguide/code-samples.run-query.html#code-samples.run-query.pagination
queryInput := ×treamquery.QueryInput{
QueryString: build_query(unloadQuery),
}
err := querySvc.QueryPages(queryInput,
func(page *timestreamquery.QueryOutput, lastPage bool) bool {
if (lastPage) {
var response = parseQueryResult(page)
var unloadFiles = getManifestAndMetadataFiles(s3Svc, response)
displayColumns(unloadFiles, unloadQuery.Partitioned_by)
displayResults(s3Svc, unloadFiles)
}
return true
})
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
}
// Utility that helps to construct UNLOAD query
type UnloadQuery struct {
Query string
Partitioned_by []string
Format string
S3Location string
ResultPrefix string
Compression string
}
func build_query(unload_query UnloadQuery) *string {
var query_results_s3_path = "'s3://" + unload_query.S3Location + "/" + unload_query.ResultPrefix + "/'"
var query = "UNLOAD(" + unload_query.Query + ") TO " + query_results_s3_path + " WITH ( "
if (len(unload_query.Partitioned_by) > 0) {
query = query + "partitioned_by=ARRAY["
for i, column := range unload_query.Partitioned_by {
if i == 0 {
query = query + "'" + column + "'"
} else {
query = query + ",'" + column + "'"
}
}
query = query + "],"
}
query = query + " format='" + unload_query.Format + "', "
query = query + " compression='" + unload_query.Compression + "')"
fmt.Println(query)
return aws.String(query)
}
```
------
#### [ Python ]
```
# When you have a SELECT like below
QUERY_1 = "SELECT user_id, ip_address, event, session_id, measure_name, time, query, quantity, product_id, channel FROM "
+ database_name + "." + table_name + " WHERE time BETWEEN ago(2d) AND now()"
# You can construct UNLOAD query as follows
UNLOAD_QUERY_1 = UnloadQuery(QUERY_1, "timestream-sample--", "without_partition", "CSV", "GZIP", "")
# Run UNLOAD query (Similar to how you run SELECT query)
# https://docs.aws.amazon.com/timestream/latest/developerguide/code-samples.run-query.html#code-samples.run-query.pagination
def run_query(self, query_string):
try:
page_iterator = self.paginator.paginate(QueryString=UNLOAD_QUERY_1)
except Exception as err:
print("Exception while running query:", err)
# Utility that helps to construct UNLOAD query
class UnloadQuery:
def __init__(self, query, s3_bucket_location, results_prefix, format, compression , partition_by):
self.query = query
self.s3_bucket_location = s3_bucket_location
self.results_prefix = results_prefix
self.format = format
self.compression = compression
self.partition_by = partition_by
def build_query(self):
query_results_s3_path = "'s3://" + self.s3_bucket_location + "/" + self.results_prefix + "/'"
unload_query = "UNLOAD("
unload_query = unload_query + self.query
unload_query = unload_query + ") "
unload_query = unload_query + " TO " + query_results_s3_path
unload_query = unload_query + " WITH ( "
if(len(self.partition_by) > 0) :
unload_query = unload_query + " partitioned_by = ARRAY" + str(self.partition_by) + ","
unload_query = unload_query + " format='" + self.format + "', "
unload_query = unload_query + " compression='" + self.compression + "')"
return unload_query
```
------
#### [ Node.js ]
```
// When you have a SELECT like below
QUERY_1 = "SELECT user_id, ip_address, event, session_id, measure_name, time, query, quantity, product_id, channel FROM "
+ database_name + "." + table_name + " WHERE time BETWEEN ago(2d) AND now()"
// You can construct UNLOAD query as follows
UNLOAD_QUERY_1 = new UnloadQuery(QUERY_1, "timestream-sample--", "without_partition", "CSV", "GZIP", "")
// Run UNLOAD query (Similar to how you run SELECT query)
// https://docs.aws.amazon.com/timestream/latest/developerguide/code-samples.run-query.html#code-samples.run-query.pagination
async runQuery(query = UNLOAD_QUERY_1, nextToken) {
const params = new QueryCommand({
QueryString: query
});
if (nextToken) {
params.NextToken = nextToken;
}
await queryClient.send(params).then(
(response) => {
if (response.NextToken) {
runQuery(queryClient, query, response.NextToken);
} else {
await parseAndDisplayResults(response);
}
},
(err) => {
console.error("Error while querying:", err);
});
}
class UnloadQuery {
constructor(query, s3_bucket_location, results_prefix, format, compression , partition_by) {
this.query = query;
this.s3_bucket_location = s3_bucket_location
this.results_prefix = results_prefix
this.format = format
this.compression = compression
this.partition_by = partition_by
}
buildQuery() {
const query_results_s3_path = "'s3://" + this.s3_bucket_location + "/" + this.results_prefix + "/'"
let unload_query = "UNLOAD("
unload_query = unload_query + this.query
unload_query = unload_query + ") "
unload_query = unload_query + " TO " + query_results_s3_path
unload_query = unload_query + " WITH ( "
if(this.partition_by.length > 0) {
let partitionBy = ""
this.partition_by.forEach((str, i) => {
partitionBy = partitionBy + (i ? ",'" : "'") + str + "'"
})
unload_query = unload_query + " partitioned_by = ARRAY[" + partitionBy + "],"
}
unload_query = unload_query + " format='" + this.format + "', "
unload_query = unload_query + " compression='" + this.compression + "')"
return unload_query
}
}
```
------
## UNLOAD レスポンスを解析し、行数、マニフェストリンク、メタデータリンクを取得する
------
#### [ Java ]
```
// Parsing UNLOAD query response is similar to how you parse SELECT query response:
// https://docs.aws.amazon.com/timestream/latest/developerguide/code-samples.run-query.html#code-samples.run-query.parsing
// But unlike SELECT, UNLOAD only has 1 row * 3 columns outputed
// (rows, metadataFile, manifestFile) => (BIGINT, VARCHAR, VARCHAR)
public UnloadResponse parseResult(QueryResult queryResult) {
Map outputMap = new HashMap<>();
for (int i = 0; i < queryResult.getColumnInfo().size(); i++) {
outputMap.put(queryResult.getColumnInfo().get(i).getName(),
queryResult.getRows().get(0).getData().get(i).getScalarValue());
}
return new UnloadResponse(outputMap);
}
@Getter
class UnloadResponse {
private final String metadataFile;
private final String manifestFile;
private final int rows;
public UnloadResponse(Map unloadResponse) {
this.metadataFile = unloadResponse.get("metadataFile");
this.manifestFile = unloadResponse.get("manifestFile");
this.rows = Integer.parseInt(unloadResponse.get("rows"));
}
}
```
------
#### [ Java v2 ]
```
// Parsing UNLOAD query response is similar to how you parse SELECT query response:
// https://docs.aws.amazon.com/timestream/latest/developerguide/code-samples.run-query.html#code-samples.run-query.parsing
// But unlike SELECT, UNLOAD only has 1 row * 3 columns outputed
// (rows, metadataFile, manifestFile) => (BIGINT, VARCHAR, VARCHAR)
public UnloadResponse parseResult(QueryResponse queryResponse) {
Map outputMap = new HashMap<>();
for (int i = 0; i < queryResponse.columnInfo().size(); i++) {
outputMap.put(queryResponse.columnInfo().get(i).name(),
queryResponse.rows().get(0).data().get(i).scalarValue());
}
return new UnloadResponse(outputMap);
}
@Getter
class UnloadResponse {
private final String metadataFile;
private final String manifestFile;
private final int rows;
public UnloadResponse(Map unloadResponse) {
this.metadataFile = unloadResponse.get("metadataFile");
this.manifestFile = unloadResponse.get("manifestFile");
this.rows = Integer.parseInt(unloadResponse.get("rows"));
}
}
```
------
#### [ Go ]
```
// Parsing UNLOAD query response is similar to how you parse SELECT query response:
// https://docs.aws.amazon.com/timestream/latest/developerguide/code-samples.run-query.html#code-samples.run-query.parsing
// But unlike SELECT, UNLOAD only has 1 row * 3 columns outputed
// (rows, metadataFile, manifestFile) => (BIGINT, VARCHAR, VARCHAR)
func parseQueryResult(queryOutput *timestreamquery.QueryOutput) map[string]string {
var columnInfo = queryOutput.ColumnInfo;
fmt.Println("ColumnInfo", columnInfo)
fmt.Println("QueryId", queryOutput.QueryId)
fmt.Println("QueryStatus", queryOutput.QueryStatus)
return parseResponse(columnInfo, queryOutput.Rows[0])
}
func parseResponse(columnInfo []*timestreamquery.ColumnInfo, row *timestreamquery.Row) map[string]string {
var datum = row.Data
response := make(map[string]string)
for i, column := range columnInfo {
response[*column.Name] = *datum[i].ScalarValue
}
return response
}
```
------
#### [ Python ]
```
# Parsing UNLOAD query response is similar to how you parse SELECT query response:
# https://docs.aws.amazon.com/timestream/latest/developerguide/code-samples.run-query.html#code-samples.run-query.parsing
# But unlike SELECT, UNLOAD only has 1 row * 3 columns outputed
# (rows, metadataFile, manifestFile) => (BIGINT, VARCHAR, VARCHAR)
for page in page_iterator:
last_page = page
response = self._parse_unload_query_result(last_page)
def _parse_unload_query_result(self, query_result):
column_info = query_result['ColumnInfo']
print("ColumnInfo: %s" % column_info)
print("QueryId: %s" % query_result['QueryId'])
print("QueryStatus:%s" % query_result['QueryStatus'])
return self.parse_unload_response(column_info, query_result['Rows'][0])
def parse_unload_response(self, column_info, row):
response = {}
data = row['Data']
for i, column in enumerate(column_info):
response[column['Name']] = data[i]['ScalarValue']
print("Rows: %s" % response['rows'])
print("Metadata File location: %s" % response['metadataFile'])
print("Manifest File location: %s" % response['manifestFile'])
return response
```
------
#### [ Node.js ]
```
# Parsing UNLOAD query response is similar to how you parse SELECT query response:
# https://docs.aws.amazon.com/timestream/latest/developerguide/code-samples.run-query.html#code-samples.run-query.parsing
# But unlike SELECT, UNLOAD only has 1 row * 3 columns outputed
# (rows, metadataFile, manifestFile) => (BIGINT, VARCHAR, VARCHAR)
async parseAndDisplayResults(data, query) {
const columnInfo = data['ColumnInfo'];
console.log("ColumnInfo:", columnInfo)
console.log("QueryId: %s", data['QueryId'])
console.log("QueryStatus:", data['QueryStatus'])
await this.parseResponse(columnInfo, data['Rows'][0], query)
}
async parseResponse(columnInfo, row, query) {
let response = {}
const data = row['Data']
columnInfo.forEach((column, i) => {
response[column['Name']] = data[i]['ScalarValue']
})
console.log("Manifest file", response['manifestFile']);
console.log("Metadata file", response['metadataFile']);
return response
}
```
------
## マニフェストコンテンツの読み取りと解析
------
#### [ Java ]
```
// Read and parse manifest content
public UnloadManifest getUnloadManifest(UnloadResponse unloadResponse) throws IOException {
AmazonS3URI s3URI = new AmazonS3URI(unloadResponse.getManifestFile());
S3Object s3Object = s3Client.getObject(s3URI.getBucket(), s3URI.getKey());
String manifestFileContent = new String(IOUtils.toByteArray(s3Object.getObjectContent()), StandardCharsets.UTF_8);
return new Gson().fromJson(manifestFileContent, UnloadManifest.class);
}
class UnloadManifest {
@Getter
public class FileMetadata {
long content_length_in_bytes;
long row_count;
}
@Getter
public class ResultFile {
String url;
FileMetadata file_metadata;
}
@Getter
public class QueryMetadata {
long total_content_length_in_bytes;
long total_row_count;
String result_format;
String result_version;
}
@Getter
public class Author {
String name;
String manifest_file_version;
}
@Getter
private List result_files;
@Getter
private QueryMetadata query_metadata;
@Getter
private Author author;
}
```
------
#### [ Java v2 ]
```
// Read and parse manifest content
public UnloadManifest getUnloadManifest(UnloadResponse unloadResponse) throws URISyntaxException {
// Space needs to encoded to use S3 parseUri function
S3Uri s3Uri = s3Utilities.parseUri(URI.create(unloadResponse.getManifestFile().replace(" ", "%20")));
ResponseBytes objectBytes = s3Client.getObjectAsBytes(GetObjectRequest.builder()
.bucket(s3Uri.bucket().orElseThrow(() -> new URISyntaxException(unloadResponse.getManifestFile(), "Invalid S3 URI")))
.key(s3Uri.key().orElseThrow(() -> new URISyntaxException(unloadResponse.getManifestFile(), "Invalid S3 URI")))
.build());
String manifestFileContent = new String(objectBytes.asByteArray(), StandardCharsets.UTF_8);
return new Gson().fromJson(manifestFileContent, UnloadManifest.class);
}
class UnloadManifest {
@Getter
public class FileMetadata {
long content_length_in_bytes;
long row_count;
}
@Getter
public class ResultFile {
String url;
FileMetadata file_metadata;
}
@Getter
public class QueryMetadata {
long total_content_length_in_bytes;
long total_row_count;
String result_format;
String result_version;
}
@Getter
public class Author {
String name;
String manifest_file_version;
}
@Getter
private List result_files;
@Getter
private QueryMetadata query_metadata;
@Getter
private Author author;
}
```
------
#### [ Go ]
```
// Read and parse manifest content
func getManifestFile(s3Svc *s3.S3, response map[string]string) Manifest {
var manifestBuf = getObject(s3Svc, response["manifestFile"])
var manifest Manifest
json.Unmarshal(manifestBuf.Bytes(), &manifest)
return manifest
}
func getObject(s3Svc *s3.S3, s3Uri string) *bytes.Buffer {
u,_ := url.Parse(s3Uri)
getObjectInput := &s3.GetObjectInput{
Key: aws.String(u.Path),
Bucket: aws.String(u.Host),
}
getObjectOutput, err := s3Svc.GetObject(getObjectInput)
if err != nil {
fmt.Println("Error: %s\n", err.Error())
}
buf := new(bytes.Buffer)
buf.ReadFrom(getObjectOutput.Body)
return buf
}
// Unload's Manifest structure
type Manifest struct {
Author interface{}
Query_metadata map[string]any
Result_files []struct {
File_metadata interface{}
Url string
}
}}
```
------
#### [ Python ]
```
def __get_manifest_file(self, response):
manifest = self.get_object(response['manifestFile']).read().decode('utf-8')
parsed_manifest = json.loads(manifest)
print("Manifest contents: \n%s" % parsed_manifest)
def get_object(self, uri):
try:
bucket, key = uri.replace("s3://", "").split("/", 1)
s3_client = boto3.client('s3', region_name=)
response = s3_client.get_object(Bucket=bucket, Key=key)
return response['Body']
except Exception as err:
print("Failed to get the object for URI:", uri)
raise err
```
------
#### [ Node.js ]
```
// Read and parse manifest content
async getManifestFile(response) {
let manifest;
await this.getS3Object(response['manifestFile']).then(
(data) => {
manifest = JSON.parse(data);
}
);
return manifest;
}
async getS3Object(uri) {
const {bucketName, key} = this.getBucketAndKey(uri);
const params = new GetObjectCommand({
Bucket: bucketName,
Key: key
})
const response = await this.s3Client.send(params);
return await response.Body.transformToString();
}
getBucketAndKey(uri) {
const [bucketName] = uri.replace("s3://", "").split("/", 1);
const key = uri.replace("s3://", "").split('/').slice(1).join('/');
return {bucketName, key};
}
```
------
## メタデータコンテンツの読み取りと解析
------
#### [ Java ]
```
// Read and parse metadata content
public UnloadMetadata getUnloadMetadata(UnloadResponse unloadResponse) throws IOException {
AmazonS3URI s3URI = new AmazonS3URI(unloadResponse.getMetadataFile());
S3Object s3Object = s3Client.getObject(s3URI.getBucket(), s3URI.getKey());
String metadataFileContent = new String(IOUtils.toByteArray(s3Object.getObjectContent()), StandardCharsets.UTF_8);
final Gson gson = new GsonBuilder()
.setFieldNamingPolicy(FieldNamingPolicy.UPPER_CAMEL_CASE)
.create();
return gson.fromJson(metadataFileContent, UnloadMetadata.class);
}
class UnloadMetadata {
@JsonProperty("ColumnInfo")
List columnInfo;
@JsonProperty("Author")
Author author;
@Data
public class Author {
@JsonProperty("Name")
String name;
@JsonProperty("MetadataFileVersion")
String metadataFileVersion;
}
}
```
------
#### [ Java v2 ]
```
// Read and parse metadata content
public UnloadMetadata getUnloadMetadata(UnloadResponse unloadResponse) throws URISyntaxException {
// Space needs to encoded to use S3 parseUri function
S3Uri s3Uri = s3Utilities.parseUri(URI.create(unloadResponse.getMetadataFile().replace(" ", "%20")));
ResponseBytes objectBytes = s3Client.getObjectAsBytes(GetObjectRequest.builder()
.bucket(s3Uri.bucket().orElseThrow(() -> new URISyntaxException(unloadResponse.getMetadataFile(), "Invalid S3 URI")))
.key(s3Uri.key().orElseThrow(() -> new URISyntaxException(unloadResponse.getMetadataFile(), "Invalid S3 URI")))
.build());
String metadataFileContent = new String(objectBytes.asByteArray(), StandardCharsets.UTF_8);
final Gson gson = new GsonBuilder()
.setFieldNamingPolicy(FieldNamingPolicy.UPPER_CAMEL_CASE)
.create();
return gson.fromJson(metadataFileContent, UnloadMetadata.class);
}
class UnloadMetadata {
@JsonProperty("ColumnInfo")
List columnInfo;
@JsonProperty("Author")
Author author;
@Data
public class Author {
@JsonProperty("Name")
String name;
@JsonProperty("MetadataFileVersion")
String metadataFileVersion;
}
}
```
------
#### [ Go ]
```
// Read and parse metadata content
func getMetadataFile(s3Svc *s3.S3, response map[string]string) Metadata {
var metadataBuf = getObject(s3Svc, response["metadataFile"])
var metadata Metadata
json.Unmarshal(metadataBuf.Bytes(), &metadata)
return metadata
}
func getObject(s3Svc *s3.S3, s3Uri string) *bytes.Buffer {
u,_ := url.Parse(s3Uri)
getObjectInput := &s3.GetObjectInput{
Key: aws.String(u.Path),
Bucket: aws.String(u.Host),
}
getObjectOutput, err := s3Svc.GetObject(getObjectInput)
if err != nil {
fmt.Println("Error: %s\n", err.Error())
}
buf := new(bytes.Buffer)
buf.ReadFrom(getObjectOutput.Body)
return buf
}
// Unload's Metadata structure
type Metadata struct {
Author interface{}
ColumnInfo []struct {
Name string
Type map[string]string
}
}
```
------
#### [ Python ]
```
def __get_metadata_file(self, response):
metadata = self.get_object(response['metadataFile']).read().decode('utf-8')
parsed_metadata = json.loads(metadata)
print("Metadata contents: \n%s" % parsed_metadata)
def get_object(self, uri):
try:
bucket, key = uri.replace("s3://", "").split("/", 1)
s3_client = boto3.client('s3', region_name=)
response = s3_client.get_object(Bucket=bucket, Key=key)
return response['Body']
except Exception as err:
print("Failed to get the object for URI:", uri)
raise err
```
------
#### [ Node.js ]
```
// Read and parse metadata content
async getMetadataFile(response) {
let metadata;
await this.getS3Object(response['metadataFile']).then(
(data) => {
metadata = JSON.parse(data);
}
);
return metadata;
}
async getS3Object(uri) {
const {bucketName, key} = this.getBucketAndKey(uri);
const params = new GetObjectCommand({
Bucket: bucketName,
Key: key
})
const response = await this.s3Client.send(params);
return await response.Body.transformToString();
}
getBucketAndKey(uri) {
const [bucketName] = uri.replace("s3://", "").split("/", 1);
const key = uri.replace("s3://", "").split('/').slice(1).join('/');
return {bucketName, key};
}
```
------
# クエリをキャンセルする
次のコードスニペットを使用してクエリをキャンセルできます。
**注記**
これらのコードスニペットは、[GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) の完全なサンプルアプリケーションに基づいています。サンプルアプリケーションの開始の詳細については、「[サンプルアプリケーション](sample-apps.md)」を参照してください。
------
#### [ Java ]
```
public void cancelQuery() {
System.out.println("Starting query: " + SELECT_ALL_QUERY);
QueryRequest queryRequest = new QueryRequest();
queryRequest.setQueryString(SELECT_ALL_QUERY);
QueryResult queryResult = queryClient.query(queryRequest);
System.out.println("Cancelling the query: " + SELECT_ALL_QUERY);
final CancelQueryRequest cancelQueryRequest = new CancelQueryRequest();
cancelQueryRequest.setQueryId(queryResult.getQueryId());
try {
queryClient.cancelQuery(cancelQueryRequest);
System.out.println("Query has been successfully cancelled");
} catch (Exception e) {
System.out.println("Could not cancel the query: " + SELECT_ALL_QUERY + " = " + e);
}
}
```
------
#### [ Java v2 ]
```
public void cancelQuery() {
System.out.println("Starting query: " + SELECT_ALL_QUERY);
QueryRequest queryRequest = QueryRequest.builder().queryString(SELECT_ALL_QUERY).build();
QueryResponse queryResponse = timestreamQueryClient.query(queryRequest);
System.out.println("Cancelling the query: " + SELECT_ALL_QUERY);
final CancelQueryRequest cancelQueryRequest = CancelQueryRequest.builder()
.queryId(queryResponse.queryId()).build();
try {
timestreamQueryClient.cancelQuery(cancelQueryRequest);
System.out.println("Query has been successfully cancelled");
} catch (Exception e) {
System.out.println("Could not cancel the query: " + SELECT_ALL_QUERY + " = " + e);
}
}
```
------
#### [ Go ]
```
cancelQueryInput := ×treamquery.CancelQueryInput{
QueryId: aws.String(*queryOutput.QueryId),
}
fmt.Println("Submitting cancellation for the query")
fmt.Println(cancelQueryInput)
// submit the query
cancelQueryOutput, err := querySvc.CancelQuery(cancelQueryInput)
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Query has been cancelled successfully")
fmt.Println(cancelQueryOutput)
}
```
------
#### [ Python ]
```
def cancel_query(self):
print("Starting query: " + self.SELECT_ALL)
result = self.client.query(QueryString=self.SELECT_ALL)
print("Cancelling query: " + self.SELECT_ALL)
try:
self.client.cancel_query(QueryId=result['QueryId'])
print("Query has been successfully cancelled")
except Exception as err:
print("Cancelling query failed:", err)
```
------
#### [ Node.js ]
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/tree/mainline/sample_apps/js)に基づいています。
```
async function tryCancelQuery() {
const params = {
QueryString: SELECT_ALL_QUERY
};
console.log(`Running query: ${SELECT_ALL_QUERY}`);
await queryClient.query(params).promise()
.then(
async (response) => {
await cancelQuery(response.QueryId);
},
(err) => {
console.error("Error while executing select all query:", err);
});
}
async function cancelQuery(queryId) {
const cancelParams = {
QueryId: queryId
};
console.log(`Sending cancellation for query: ${SELECT_ALL_QUERY}`);
await queryClient.cancelQuery(cancelParams).promise()
.then(
(response) => {
console.log("Query has been cancelled successfully");
},
(err) => {
console.error("Error while cancelling select all:", err);
});
}
```
------
#### [ .NET ]
```
public async Task CancelQuery()
{
Console.WriteLine("Starting query: " + SELECT_ALL_QUERY);
QueryRequest queryRequest = new QueryRequest();
queryRequest.QueryString = SELECT_ALL_QUERY;
QueryResponse queryResponse = await queryClient.QueryAsync(queryRequest);
Console.WriteLine("Cancelling query: " + SELECT_ALL_QUERY);
CancelQueryRequest cancelQueryRequest = new CancelQueryRequest();
cancelQueryRequest.QueryId = queryResponse.QueryId;
try
{
await queryClient.CancelQueryAsync(cancelQueryRequest);
Console.WriteLine("Query has been successfully cancelled.");
} catch(Exception e)
{
Console.WriteLine("Could not cancel the query: " + SELECT_ALL_QUERY + " = " + e);
}
}
```
------
# バッチロードタスクを作成する
次のコードスニペットを使用して、バッチロードタスクを作成できます。
------
#### [ Java ]
```
package com.example.tryit;
import java.util.Arrays;
import software.amazon.awssdk.services.timestreamwrite.model.CreateBatchLoadTaskRequest;
import software.amazon.awssdk.services.timestreamwrite.model.CreateBatchLoadTaskResponse;
import software.amazon.awssdk.services.timestreamwrite.model.DataModel;
import software.amazon.awssdk.services.timestreamwrite.model.DataModelConfiguration;
import software.amazon.awssdk.services.timestreamwrite.model.DataSourceConfiguration;
import software.amazon.awssdk.services.timestreamwrite.model.DataSourceS3Configuration;
import software.amazon.awssdk.services.timestreamwrite.model.DimensionMapping;
import software.amazon.awssdk.services.timestreamwrite.model.MultiMeasureAttributeMapping;
import software.amazon.awssdk.services.timestreamwrite.model.MultiMeasureMappings;
import software.amazon.awssdk.services.timestreamwrite.model.ReportConfiguration;
import software.amazon.awssdk.services.timestreamwrite.model.ReportS3Configuration;
import software.amazon.awssdk.services.timestreamwrite.model.ScalarMeasureValueType;
import software.amazon.awssdk.services.timestreamwrite.model.TimeUnit;
import software.amazon.awssdk.services.timestreamwrite.TimestreamWriteClient;
public class BatchLoadExample {
public static final String DATABASE_NAME = ;
public static final String TABLE_NAME = ;
public static final String INPUT_BUCKET = ;
public static final String INPUT_OBJECT_KEY_PREFIX = ;
public static final String REPORT_BUCKET = ;
public static final long HT_TTL_HOURS = 24L;
public static final long CT_TTL_DAYS = 7L;
TimestreamWriteClient amazonTimestreamWrite;
public BatchLoadExample(TimestreamWriteClient client) {
this.amazonTimestreamWrite = client;
}
public String createBatchLoadTask() {
System.out.println("Creating batch load task");
CreateBatchLoadTaskRequest request = CreateBatchLoadTaskRequest.builder()
.dataModelConfiguration(DataModelConfiguration.builder()
.dataModel(DataModel.builder()
.timeColumn("timestamp")
.timeUnit(TimeUnit.SECONDS)
.dimensionMappings(Arrays.asList(
DimensionMapping.builder()
.sourceColumn("vehicle")
.build(),
DimensionMapping.builder()
.sourceColumn("registration")
.destinationColumn("license")
.build()))
.multiMeasureMappings(MultiMeasureMappings.builder()
.targetMultiMeasureName("mva_measure_name")
.multiMeasureAttributeMappings(Arrays.asList(
MultiMeasureAttributeMapping.builder()
.sourceColumn("wgt")
.targetMultiMeasureAttributeName("weight")
.measureValueType(ScalarMeasureValueType.DOUBLE)
.build(),
MultiMeasureAttributeMapping.builder()
.sourceColumn("spd")
.targetMultiMeasureAttributeName("speed")
.measureValueType(ScalarMeasureValueType.DOUBLE)
.build(),
MultiMeasureAttributeMapping.builder()
.sourceColumn("fuel")
.measureValueType(ScalarMeasureValueType.DOUBLE)
.build(),
MultiMeasureAttributeMapping.builder()
.sourceColumn("miles")
.measureValueType(ScalarMeasureValueType.DOUBLE)
.build()))
.build())
.build())
.build())
.dataSourceConfiguration(DataSourceConfiguration.builder()
.dataSourceS3Configuration(
DataSourceS3Configuration.builder()
.bucketName(INPUT_BUCKET)
.objectKeyPrefix(INPUT_OBJECT_KEY_PREFIX)
.build())
.dataFormat("CSV")
.build())
.reportConfiguration(ReportConfiguration.builder()
.reportS3Configuration(ReportS3Configuration.builder()
.bucketName(REPORT_BUCKET)
.build())
.build())
.targetDatabaseName(DATABASE_NAME)
.targetTableName(TABLE_NAME)
.build();
try {
final CreateBatchLoadTaskResponse createBatchLoadTaskResponse = amazonTimestreamWrite.createBatchLoadTask(request);
String taskId = createBatchLoadTaskResponse.taskId();
System.out.println("Successfully created batch load task: " + taskId);
return taskId;
} catch (Exception e) {
System.out.println("Failed to create batch load task: " + e);
throw e;
}
}
}
```
------
#### [ Go ]
```
package main
import (
"fmt"
"context"
"log"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/config"
"github.com/aws/aws-sdk-go-v2/service/timestreamwrite"
"github.com/aws/aws-sdk-go-v2/service/timestreamwrite/types"
)
func main() {
customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{})(aws.Endpoint, error) {
if service == timestreamwrite.ServiceID && region == "us-west-2" {
return aws.Endpoint{
PartitionID: "aws",
URL: ,
SigningRegion: "us-west-2",
}, nil
}
return aws.Endpoint{}, & aws.EndpointNotFoundError{}
})
cfg, err := config.LoadDefaultConfig(context.TODO(), config.WithEndpointResolverWithOptions(customResolver), config.WithRegion("us-west-2"))
if err != nil {
log.Fatalf("failed to load configuration, %v", err)
}
client := timestreamwrite.NewFromConfig(cfg)
response, err := client.CreateBatchLoadTask(context.TODO(), & timestreamwrite.CreateBatchLoadTaskInput{
TargetDatabaseName: aws.String("BatchLoadExampleDatabase"),
TargetTableName: aws.String("BatchLoadExampleTable"),
RecordVersion: aws.Int64(1),
DataModelConfiguration: & types.DataModelConfiguration{
DataModel: & types.DataModel{
TimeColumn: aws.String("timestamp"),
TimeUnit: types.TimeUnitMilliseconds,
DimensionMappings: []types.DimensionMapping{
{
SourceColumn: aws.String("registration"),
DestinationColumn: aws.String("license"),
},
},
MultiMeasureMappings: & types.MultiMeasureMappings{
TargetMultiMeasureName: aws.String("mva_measure_name"),
MultiMeasureAttributeMappings: []types.MultiMeasureAttributeMapping{
{
SourceColumn: aws.String("wgt"),
TargetMultiMeasureAttributeName: aws.String("weight"),
MeasureValueType: types.ScalarMeasureValueTypeDouble,
},
{
SourceColumn: aws.String("spd"),
TargetMultiMeasureAttributeName: aws.String("speed"),
MeasureValueType: types.ScalarMeasureValueTypeDouble,
},
{
SourceColumn: aws.String("fuel_consumption"),
TargetMultiMeasureAttributeName: aws.String("fuel"),
MeasureValueType: types.ScalarMeasureValueTypeDouble,
},
},
},
},
},
DataSourceConfiguration: & types.DataSourceConfiguration{
DataSourceS3Configuration: & types.DataSourceS3Configuration{
BucketName: aws.String("test-batch-load-west-2"),
ObjectKeyPrefix: aws.String("sample.csv"),
},
DataFormat: types.BatchLoadDataFormatCsv,
},
ReportConfiguration: & types.ReportConfiguration{
ReportS3Configuration: & types.ReportS3Configuration{
BucketName: aws.String("test-batch-load-report-west-2"),
EncryptionOption: types.S3EncryptionOptionSseS3,
},
},
})
fmt.Println(aws.ToString(response.TaskId))
}
```
------
#### [ Python ]
```
import boto3
from botocore.config import Config
INGEST_ENDPOINT = ""
REGION = "us-west-2"
HT_TTL_HOURS = 24
CT_TTL_DAYS = 7
DATABASE_NAME = ""
TABLE_NAME = ""
INPUT_BUCKET_NAME = ""
INPUT_OBJECT_KEY_PREFIX = ""
REPORT_BUCKET_NAME = ""
def create_batch_load_task(client, database_name, table_name, input_bucket_name, input_object_key_prefix, report_bucket_name):
try:
result = client.create_batch_load_task(TargetDatabaseName=database_name, TargetTableName=table_name,
DataModelConfiguration={"DataModel": {
"TimeColumn": "timestamp",
"TimeUnit": "SECONDS",
"DimensionMappings": [
{
"SourceColumn": "vehicle"
},
{
"SourceColumn": "registration",
"DestinationColumn": "license"
}
],
"MultiMeasureMappings": {
"TargetMultiMeasureName": "metrics",
"MultiMeasureAttributeMappings": [
{
"SourceColumn": "wgt",
"MeasureValueType": "DOUBLE"
},
{
"SourceColumn": "spd",
"MeasureValueType": "DOUBLE"
},
{
"SourceColumn": "fuel_consumption",
"TargetMultiMeasureAttributeName": "fuel",
"MeasureValueType": "DOUBLE"
},
{
"SourceColumn": "miles",
"MeasureValueType": "DOUBLE"
}
]}
}
},
DataSourceConfiguration={
"DataSourceS3Configuration": {
"BucketName": input_bucket_name,
"ObjectKeyPrefix": input_object_key_prefix
},
"DataFormat": "CSV"
},
ReportConfiguration={
"ReportS3Configuration": {
"BucketName": report_bucket_name,
"EncryptionOption": "SSE_S3"
}
}
)
task_id = result["TaskId"]
print("Successfully created batch load task: ", task_id)
return task_id
except Exception as err:
print("Create batch load task job failed:", err)
return None
if __name__ == '__main__':
session = boto3.Session()
write_client = session.client('timestream-write',
endpoint_url=INGEST_ENDPOINT, region_name=REGION,
config=Config(read_timeout=20, max_pool_connections=5000, retries={'max_attempts': 10}))
task_id = create_batch_load_task(write_client, DATABASE_NAME, TABLE_NAME,
INPUT_BUCKET_NAME, INPUT_OBJECT_KEY_PREFIX, REPORT_BUCKET_NAME)
```
------
#### [ Node.js ]
以下のスニペットでは、AWS SDK for JavaScript v3 を使用しています。クライアントのインストール方法と使用の詳細については、「[Timestream Write クライアント – AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html)」を参照してください。
API の詳細については、「[クラス CreateBatchLoadCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/createbatchloadtaskcommand.html)」および「[CreateBatchLoadTask](https://docs.aws.amazon.com/timestream/latest/developerguide/API_CreateBatchLoadTask.html)」を参照してください。
```
import { TimestreamWriteClient, CreateBatchLoadTaskCommand } from "@aws-sdk/client-timestream-write";
const writeClient = new TimestreamWriteClient({ region: "us-west-2", endpoint: "https://gamma-ingest-cell3.timestream.us-west-2.amazonaws.com" });
const params = {
TargetDatabaseName: "BatchLoadExampleDatabase",
TargetTableName: "BatchLoadExampleTable",
RecordVersion: 1,
DataModelConfiguration: {
DataModel: {
TimeColumn: "timestamp",
TimeUnit: "MILLISECONDS",
DimensionMappings: [
{
SourceColumn: "registration",
DestinationColumn: "license"
}
],
MultiMeasureMappings: {
TargetMultiMeasureName: "mva_measure_name",
MultiMeasureAttributeMappings: [
{
SourceColumn: "wgt",
TargetMultiMeasureAttributeName: "weight",
MeasureValueType: "DOUBLE"
},
{
SourceColumn: "spd",
TargetMultiMeasureAttributeName: "speed",
MeasureValueType: "DOUBLE"
},
{
SourceColumn: "fuel_consumption",
TargetMultiMeasureAttributeName: "fuel",
MeasureValueType: "DOUBLE"
}
]
}
}
},
DataSourceConfiguration: {
DataSourceS3Configuration: {
BucketName: "test-batch-load-west-2",
ObjectKeyPrefix: "sample.csv"
},
DataFormat: "CSV"
},
ReportConfiguration: {
ReportS3Configuration: {
BucketName: "test-batch-load-report-west-2",
EncryptionOption: "SSE_S3"
}
}
};
const command = new CreateBatchLoadTaskCommand(params);
try {
const data = await writeClient.send(command);
console.log(`Created batch load task ` + data.TaskId);
} catch (error) {
console.log("Error creating table. ", error);
throw error;
}
```
------
#### [ .NET ]
```
using System;
using System.IO;
using System.Collections.Generic;
using Amazon.TimestreamWrite;
using Amazon.TimestreamWrite.Model;
using System.Threading.Tasks;
namespace TimestreamDotNetSample
{
public class CreateBatchLoadTaskExample
{
public const string DATABASE_NAME = "";
public const string TABLE_NAME = "";
public const string INPUT_BUCKET = "";
public const string INPUT_OBJECT_KEY_PREFIX = "";
public const string REPORT_BUCKET = "";
public const long HT_TTL_HOURS = 24L;
public const long CT_TTL_DAYS = 7L;
private readonly AmazonTimestreamWriteClient writeClient;
public CreateBatchLoadTaskExample(AmazonTimestreamWriteClient writeClient)
{
this.writeClient = writeClient;
}
public async Task CreateBatchLoadTask()
{
try
{
var createBatchLoadTaskRequest = new CreateBatchLoadTaskRequest
{
DataModelConfiguration = new DataModelConfiguration
{
DataModel = new DataModel
{
TimeColumn = "timestamp",
TimeUnit = TimeUnit.SECONDS,
DimensionMappings = new List()
{
new()
{
SourceColumn = "vehicle"
},
new()
{
SourceColumn = "registration",
DestinationColumn = "license"
}
},
MultiMeasureMappings = new MultiMeasureMappings
{
TargetMultiMeasureName = "mva_measure_name",
MultiMeasureAttributeMappings = new List()
{
new()
{
SourceColumn = "wgt",
TargetMultiMeasureAttributeName = "weight",
MeasureValueType = ScalarMeasureValueType.DOUBLE
},
new()
{
SourceColumn = "spd",
TargetMultiMeasureAttributeName = "speed",
MeasureValueType = ScalarMeasureValueType.DOUBLE
},
new()
{
SourceColumn = "fuel",
TargetMultiMeasureAttributeName = "fuel",
MeasureValueType = ScalarMeasureValueType.DOUBLE
},
new()
{
SourceColumn = "miles",
TargetMultiMeasureAttributeName = "miles",
MeasureValueType = ScalarMeasureValueType.DOUBLE
}
}
}
}
},
DataSourceConfiguration = new DataSourceConfiguration
{
DataSourceS3Configuration = new DataSourceS3Configuration
{
BucketName = INPUT_BUCKET,
ObjectKeyPrefix = INPUT_OBJECT_KEY_PREFIX
},
DataFormat = "CSV"
},
ReportConfiguration = new ReportConfiguration
{
ReportS3Configuration = new ReportS3Configuration
{
BucketName = REPORT_BUCKET
}
},
TargetDatabaseName = DATABASE_NAME,
TargetTableName = TABLE_NAME
};
CreateBatchLoadTaskResponse response = await writeClient.CreateBatchLoadTaskAsync(createBatchLoadTaskRequest);
Console.WriteLine($"Task created: " + response.TaskId);
}
catch (Exception e)
{
Console.WriteLine("Create batch load task failed:" + e.ToString());
}
}
}
}
```
```
using Amazon.TimestreamWrite;
using Amazon.TimestreamWrite.Model;
using Amazon;
using Amazon.TimestreamQuery;
using System.Threading.Tasks;
using System;
using CommandLine;
static class Constants
{
}
namespace TimestreamDotNetSample
{
class MainClass
{
public class Options
{
}
public static void Main(string[] args)
{
Parser.Default.ParseArguments(args)
.WithParsed(o => {
MainAsync().GetAwaiter().GetResult();
});
}
static async Task MainAsync()
{
var writeClientConfig = new AmazonTimestreamWriteConfig
{
ServiceURL = "",
Timeout = TimeSpan.FromSeconds(20),
MaxErrorRetry = 10
};
var writeClient = new AmazonTimestreamWriteClient(writeClientConfig);
var example = new CreateBatchLoadTaskExample(writeClient);
await example.CreateBatchLoadTask();
}
}
}
```
------
# バッチロードタスクを記述する
次のコードスニペットを使用して、バッチロードタスクを記述できます。
------
#### [ Java ]
```
public void describeBatchLoadTask(String taskId) {
final DescribeBatchLoadTaskResponse batchLoadTaskResponse = amazonTimestreamWrite
.describeBatchLoadTask(DescribeBatchLoadTaskRequest.builder()
.taskId(taskId)
.build());
System.out.println("Task id: " + batchLoadTaskResponse.batchLoadTaskDescription().taskId());
System.out.println("Status: " + batchLoadTaskResponse.batchLoadTaskDescription().taskStatusAsString());
System.out.println("Records processed: "
+ batchLoadTaskResponse.batchLoadTaskDescription().progressReport().recordsProcessed());
}
```
------
#### [ Go ]
```
package main
import (
"fmt"
"context"
"log"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/config"
"github.com/aws/aws-sdk-go-v2/service/timestreamwrite"
)
func main() {
customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {
if service == timestreamwrite.ServiceID && region == "us-west-2" {
return aws.Endpoint{
PartitionID: "aws",
URL: ,
SigningRegion: "us-west-2",
}, nil
}
return aws.Endpoint{}, &aws.EndpointNotFoundError{}
})
cfg, err := config.LoadDefaultConfig(context.TODO(), config.WithEndpointResolverWithOptions(customResolver), config.WithRegion("us-west-2"))
if err != nil {
log.Fatalf("failed to load configuration, %v", err)
}
client := timestreamwrite.NewFromConfig(cfg)
response, err := client.DescribeBatchLoadTask(context.TODO(), ×treamwrite.DescribeBatchLoadTaskInput{
TaskId: aws.String(""),
})
fmt.Println(aws.ToString(response.BatchLoadTaskDescription.TaskId))
}
```
------
#### [ Python ]
```
import boto3
from botocore.config import Config
INGEST_ENDPOINT=""
REGION="us-west-2"
HT_TTL_HOURS = 24
CT_TTL_DAYS = 7
TASK_ID = ""
def describe_batch_load_task(client, task_id):
try:
result = client.describe_batch_load_task(TaskId=task_id)
print("Successfully described batch load task: ", result)
except Exception as err:
print("Describe batch load task job failed:", err)
if __name__ == '__main__':
session = boto3.Session()
write_client = session.client('timestream-write', \
endpoint_url=INGEST_ENDPOINT, region_name=REGION, \
config=Config(read_timeout=20, max_pool_connections = 5000, retries={'max_attempts': 10}))
describe_batch_load_task(write_client, TASK_ID)
```
------
#### [ Node.js ]
以下のスニペットでは、AWS SDK for JavaScript v3 を使用しています。クライアントのインストール方法と使用の詳細については、「[Timestream Write クライアント – AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html)」を参照してください。
API の詳細については、「[クラス DescribeBatchLoadCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/describebatchloadtaskcommand.html)」および「[DescribeBatchLoadTask](https://docs.aws.amazon.com/timestream/latest/developerguide/API_DescribeBatchLoadTask.html)」を参照してください。
```
import { TimestreamWriteClient, DescribeBatchLoadTaskCommand } from "@aws-sdk/client-timestream-write";
const writeClient = new TimestreamWriteClient({ region: "", endpoint: "" });
const params = {
TaskId: ""
};
const command = new DescribeBatchLoadTaskCommand(params);
try {
const data = await writeClient.send(command);
console.log(`Batch load task has id ` + data.BatchLoadTaskDescription.TaskId);
} catch (error) {
if (error.code === 'ResourceNotFoundException') {
console.log("Batch load task doesn't exist.");
} else {
console.log("Describe batch load task failed.", error);
throw error;
}
}
```
------
#### [ .NET ]
```
using System;
using System.IO;
using System.Collections.Generic;
using Amazon.TimestreamWrite;
using Amazon.TimestreamWrite.Model;
using System.Threading.Tasks;
namespace TimestreamDotNetSample
{
public class DescribeBatchLoadTaskExample
{
private readonly AmazonTimestreamWriteClient writeClient;
public DescribeBatchLoadTaskExample(AmazonTimestreamWriteClient writeClient)
{
this.writeClient = writeClient;
}
public async Task DescribeBatchLoadTask(String taskId)
{
try
{
var describeBatchLoadTaskRequest = new DescribeBatchLoadTaskRequest
{
TaskId = taskId
};
DescribeBatchLoadTaskResponse response = await writeClient.DescribeBatchLoadTaskAsync(describeBatchLoadTaskRequest);
Console.WriteLine($"Task has id:{response.BatchLoadTaskDescription.TaskId}");
}
catch (ResourceNotFoundException)
{
Console.WriteLine("Batch load task does not exist.");
}
catch (Exception e)
{
Console.WriteLine("Describe batch load task failed:" + e.ToString());
}
}
}
}
```
```
using Amazon.TimestreamWrite;
using Amazon.TimestreamWrite.Model;
using Amazon;
using Amazon.TimestreamQuery;
using System.Threading.Tasks;
using System;
using CommandLine;
static class Constants
{
}
namespace TimestreamDotNetSample
{
class MainClass
{
public class Options
{
}
public static void Main(string[] args)
{
Parser.Default.ParseArguments(args)
.WithParsed(o => {
MainAsync().GetAwaiter().GetResult();
});
}
static async Task MainAsync()
{
var writeClientConfig = new AmazonTimestreamWriteConfig
{
ServiceURL = "",
Timeout = TimeSpan.FromSeconds(20),
MaxErrorRetry = 10
};
var writeClient = new AmazonTimestreamWriteClient(writeClientConfig);
var example = new DescribeBatchLoadTaskExample(writeClient);
await example.DescribeBatchLoadTask("");
}
}
}
```
------
# バッチロードタスクを一覧表示する
次のコードスニペットを使用してバッチロードタスクを一覧表示できます。
------
#### [ Java ]
```
public void listBatchLoadTasks() {
final ListBatchLoadTasksResponse listBatchLoadTasksResponse = amazonTimestreamWrite
.listBatchLoadTasks(ListBatchLoadTasksRequest.builder()
.maxResults(15)
.build());
for (BatchLoadTask batchLoadTask : listBatchLoadTasksResponse.batchLoadTasks()) {
System.out.println(batchLoadTask.taskId());
}
}
```
------
#### [ Go ]
```
package main
import (
"fmt"
"context"
"log"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/config"
"github.com/aws/aws-sdk-go-v2/service/timestreamwrite"
)
func main() {
customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {
if service == timestreamwrite.ServiceID && region == "us-west-2" {
return aws.Endpoint{
PartitionID: "aws",
URL: ,
SigningRegion: "us-west-2",
}, nil
}
return aws.Endpoint{}, &aws.EndpointNotFoundError{}
})
cfg, err := config.LoadDefaultConfig(context.TODO(), config.WithEndpointResolverWithOptions(customResolver), config.WithRegion("us-west-2"))
if err != nil {
log.Fatalf("failed to load configuration, %v", err)
}
client := timestreamwrite.NewFromConfig(cfg)
listBatchLoadTasksMaxResult := int32(15)
response, err := client.ListBatchLoadTasks(context.TODO(), ×treamwrite.ListBatchLoadTasksInput{
MaxResults: &listBatchLoadTasksMaxResult,
})
for i, task := range response.BatchLoadTasks {
fmt.Println(i, aws.ToString(task.TaskId))
}
}
```
------
#### [ Python ]
```
import boto3
from botocore.config import Config
INGEST_ENDPOINT = ""
REGION = "us-west-2"
HT_TTL_HOURS = 24
CT_TTL_DAYS = 7
def print_batch_load_tasks(batch_load_tasks):
for batch_load_task in batch_load_tasks:
print(batch_load_task['TaskId'])
def list_batch_load_tasks(client):
print("\nListing batch load tasks")
try:
response = client.list_batch_load_tasks(MaxResults=10)
print_batch_load_tasks(response['BatchLoadTasks'])
next_token = response.get('NextToken', None)
while next_token:
response = client.list_batch_load_tasks(
NextToken=next_token, MaxResults=10)
print_batch_load_tasks(response['BatchLoadTasks'])
next_token = response.get('NextToken', None)
except Exception as err:
print("List batch load tasks failed:", err)
raise err
if __name__ == '__main__':
session = boto3.Session()
write_client = session.client('timestream-write',
endpoint_url=INGEST_ENDPOINT, region_name=REGION,
config=Config(read_timeout=20, max_pool_connections=5000, retries={'max_attempts': 10}))
list_batch_load_tasks(write_client)
```
------
#### [ Node.js ]
以下のスニペットでは、AWS SDK for JavaScript v3 を使用しています。クライアントのインストール方法と使用の詳細については、「[Timestream Write クライアント – AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html)」を参照してください。
API の詳細については、「[クラス DescribeBatchLoadCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/listbatchloadtaskscommand.html)」および「[DescribeBatchLoadTask](https://docs.aws.amazon.com/timestream/latest/developerguide/API_DescribeBatchLoadTask.html)」を参照してください。
```
import { TimestreamWriteClient, ListBatchLoadTasksCommand } from "@aws-sdk/client-timestream-write";
const writeClient = new TimestreamWriteClient({ region: "", endpoint: "" });
const params = {
MaxResults: <15>
};
const command = new ListBatchLoadTasksCommand(params);
getBatchLoadTasksList(null);
async function getBatchLoadTasksList(nextToken) {
if (nextToken) {
params.NextToken = nextToken;
}
try {
const data = await writeClient.send(command);
data.BatchLoadTasks.forEach(function (task) {
console.log(task.TaskId);
});
if (data.NextToken) {
return getBatchLoadTasksList(data.NextToken);
}
} catch (error) {
console.log("Error while listing batch load tasks", error);
}
}
```
------
#### [ .NET ]
```
using System;
using System.IO;
using System.Collections.Generic;
using Amazon.TimestreamWrite;
using Amazon.TimestreamWrite.Model;
using System.Threading.Tasks;
namespace TimestreamDotNetSample
{
public class ListBatchLoadTasksExample
{
private readonly AmazonTimestreamWriteClient writeClient;
public ListBatchLoadTasksExample(AmazonTimestreamWriteClient writeClient)
{
this.writeClient = writeClient;
}
public async Task ListBatchLoadTasks()
{
Console.WriteLine("Listing batch load tasks");
try
{
var listBatchLoadTasksRequest = new ListBatchLoadTasksRequest
{
MaxResults = 15
};
ListBatchLoadTasksResponse response = await writeClient.ListBatchLoadTasksAsync(listBatchLoadTasksRequest);
PrintBatchLoadTasks(response.BatchLoadTasks);
var nextToken = response.NextToken;
while (nextToken != null)
{
listBatchLoadTasksRequest.NextToken = nextToken;
response = await writeClient.ListBatchLoadTasksAsync(listBatchLoadTasksRequest);
PrintBatchLoadTasks(response.BatchLoadTasks);
nextToken = response.NextToken;
}
}
catch (Exception e)
{
Console.WriteLine("List batch load tasks failed:" + e.ToString());
}
}
private void PrintBatchLoadTasks(List tasks)
{
foreach (BatchLoadTask task in tasks)
Console.WriteLine($"Task:{task.TaskId}");
}
}
}
```
```
using Amazon.TimestreamWrite;
using Amazon.TimestreamWrite.Model;
using Amazon;
using Amazon.TimestreamQuery;
using System.Threading.Tasks;
using System;
using CommandLine;
static class Constants
{
}
namespace TimestreamDotNetSample
{
class MainClass
{
public class Options
{
}
public static void Main(string[] args)
{
Parser.Default.ParseArguments(args)
.WithParsed(o => {
MainAsync().GetAwaiter().GetResult();
});
}
static async Task MainAsync()
{
var writeClientConfig = new AmazonTimestreamWriteConfig
{
ServiceURL = "",
Timeout = TimeSpan.FromSeconds(20),
MaxErrorRetry = 10
};
var writeClient = new AmazonTimestreamWriteClient(writeClientConfig);
var example = new ListBatchLoadTasksExample(writeClient);
await example.ListBatchLoadTasks();
}
}
}
```
------
# バッチロードタスクを再開する
次のコードスニペットを使用してバッチロードタスクを再開できます。
------
#### [ Java ]
```
public void resumeBatchLoadTask(String taskId) {
try {
amazonTimestreamWrite
.resumeBatchLoadTask(ResumeBatchLoadTaskRequest.builder()
.taskId(taskId)
.build());
System.out.println("Successfully resumed batch load task.");
} catch (ValidationException validationException) {
System.out.println(validationException.getMessage());
}
}
```
------
#### [ Go ]
```
package main
import (
"fmt"
"context"
"log"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/config"
"github.com/aws/aws-sdk-go-v2/service/timestreamwrite"
)
func main() {
customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {
if service == timestreamwrite.ServiceID && region == "us-west-2" {
return aws.Endpoint{
PartitionID: "aws",
URL: ,
SigningRegion: "us-west-2",
}, nil
}
return aws.Endpoint{}, &aws.EndpointNotFoundError{}
})
cfg, err := config.LoadDefaultConfig(context.TODO(), config.WithEndpointResolverWithOptions(customResolver), config.WithRegion("us-west-2"))
if err != nil {
log.Fatalf("failed to load configuration, %v", err)
}
client := timestreamwrite.NewFromConfig(cfg)
response, err := client.ResumeBatchLoadTask(context.TODO(), ×treamwrite.ResumeBatchLoadTaskInput{
TaskId: aws.String("TaskId"),
})
if err != nil {
fmt.Println("Error:")
fmt.Println(err)
} else {
fmt.Println("Resume batch load task is successful")
fmt.Println(response)
}
}
```
------
#### [ Python ]
```
import boto3
from botocore.config import Config
INGEST_ENDPOINT=""
REGION="us-west-2"
HT_TTL_HOURS = 24
CT_TTL_DAYS = 7
TASK_ID = ""
def resume_batch_load_task(client, task_id):
try:
result = client.resume_batch_load_task(TaskId=task_id)
print("Successfully resumed batch load task: ", result)
except Exception as err:
print("Resume batch load task failed:", err)
if __name__ == '__main__':
session = boto3.Session()
write_client = session.client('timestream-write', \
endpoint_url=INGEST_ENDPOINT, region_name=REGION, \
config=Config(read_timeout=20, max_pool_connections = 5000, retries={'max_attempts': 10}))
resume_batch_load_task(write_client, TASK_ID)
```
------
#### [ Node.js ]
以下のスニペットでは、AWS SDK for JavaScript v3 を使用しています。クライアントのインストール方法と使用の詳細については、「[Timestream Write クライアント – AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html)」を参照してください。
API の詳細については、「[クラス CreateBatchLoadCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/describebatchloadtaskcommand.html)」および「[CreateBatchLoadTask](https://docs.aws.amazon.com/timestream/latest/developerguide/API_CreateBatchLoadTask.html)」を参照してください。
```
import { TimestreamWriteClient, ResumeBatchLoadTaskCommand } from "@aws-sdk/client-timestream-write";
const writeClient = new TimestreamWriteClient({ region: "", endpoint: "" });
const params = {
TaskId: ""
};
const command = new ResumeBatchLoadTaskCommand(params);
try {
const data = await writeClient.send(command);
console.log("Resumed batch load task");
} catch (error) {
console.log("Resume batch load task failed.", error);
throw error;
}
```
------
#### [ .NET ]
```
using System;
using System.IO;
using System.Collections.Generic;
using Amazon.TimestreamWrite;
using Amazon.TimestreamWrite.Model;
using System.Threading.Tasks;
namespace TimestreamDotNetSample
{
public class ResumeBatchLoadTaskExample
{
private readonly AmazonTimestreamWriteClient writeClient;
public ResumeBatchLoadTaskExample(AmazonTimestreamWriteClient writeClient)
{
this.writeClient = writeClient;
}
public async Task ResumeBatchLoadTask(String taskId)
{
try
{
var resumeBatchLoadTaskRequest = new ResumeBatchLoadTaskRequest
{
TaskId = taskId
};
ResumeBatchLoadTaskResponse response = await writeClient.ResumeBatchLoadTaskAsync(resumeBatchLoadTaskRequest);
Console.WriteLine("Successfully resumed batch load task.");
}
catch (ResourceNotFoundException)
{
Console.WriteLine("Batch load task does not exist.");
}
catch (Exception e)
{
Console.WriteLine("Resume batch load task failed: " + e.ToString());
}
}
}
}
```
------
# スケジュールされたクエリを作成する
次のコードスニペットを使用して、マルチメジャーマッピングでスケジュールされたクエリを作成できます。
------
#### [ Java ]
```
public static String DATABASE_NAME = "devops_sample_application";
public static String TABLE_NAME = "host_metrics_sample_application";
public static String HOSTNAME = "host-24Gju";
public static String SQ_NAME = "daily-sample";
public static String SCHEDULE_EXPRESSION = "cron(0/2 * * * ? *)";
// Find the average, p90, p95, and p99 CPU utilization for a specific EC2 host over the past 2 hours.
public static String QUERY = "SELECT region, az, hostname, BIN(time, 15s) AS binned_timestamp, " +
"ROUND(AVG(cpu_utilization), 2) AS avg_cpu_utilization, " +
"ROUND(APPROX_PERCENTILE(cpu_utilization, 0.9), 2) AS p90_cpu_utilization, " +
"ROUND(APPROX_PERCENTILE(cpu_utilization, 0.95), 2) AS p95_cpu_utilization, " +
"ROUND(APPROX_PERCENTILE(cpu_utilization, 0.99), 2) AS p99_cpu_utilization " +
"FROM " + DATABASE_NAME + "." + TABLE_NAME + " " +
"WHERE measure_name = 'metrics' " +
"AND hostname = '" + HOSTNAME + "' " +
"AND time > ago(2h) " +
"GROUP BY region, hostname, az, BIN(time, 15s) " +
"ORDER BY binned_timestamp ASC " +
"LIMIT 5";
public String createScheduledQuery(String topic_arn,
String role_arn,
String database_name,
String table_name) {
System.out.println("Creating Scheduled Query");
List> sourceColToMeasureValueTypes = Arrays.asList(
Pair.of("avg_cpu_utilization", DOUBLE),
Pair.of("p90_cpu_utilization", DOUBLE),
Pair.of("p95_cpu_utilization", DOUBLE),
Pair.of("p99_cpu_utilization", DOUBLE));
CreateScheduledQueryRequest createScheduledQueryRequest = new CreateScheduledQueryRequest()
.withName(SQ_NAME)
.withQueryString(QUERY)
.withScheduleConfiguration(new ScheduleConfiguration()
.withScheduleExpression(SCHEDULE_EXPRESSION))
.withNotificationConfiguration(new NotificationConfiguration()
.withSnsConfiguration(new SnsConfiguration()
.withTopicArn(topic_arn)))
.withTargetConfiguration(new TargetConfiguration().withTimestreamConfiguration(new TimestreamConfiguration()
.withDatabaseName(database_name)
.withTableName(table_name)
.withTimeColumn("binned_timestamp")
.withDimensionMappings(Arrays.asList(
new DimensionMapping()
.withName("region")
.withDimensionValueType("VARCHAR"),
new DimensionMapping()
.withName("az")
.withDimensionValueType("VARCHAR"),
new DimensionMapping()
.withName("hostname")
.withDimensionValueType("VARCHAR")
))
.withMultiMeasureMappings(new MultiMeasureMappings()
.withTargetMultiMeasureName("multi-metrics")
.withMultiMeasureAttributeMappings(
sourceColToMeasureValueTypes.stream()
.map(pair -> new MultiMeasureAttributeMapping()
.withMeasureValueType(pair.getValue().name())
.withSourceColumn(pair.getKey()))
.collect(Collectors.toList())))))
.withErrorReportConfiguration(new ErrorReportConfiguration()
.withS3Configuration(new S3Configuration()
.withBucketName(timestreamDependencyHelper.getS3ErrorReportBucketName())))
.withScheduledQueryExecutionRoleArn(role_arn);
try {
final CreateScheduledQueryResult createScheduledQueryResult = queryClient.createScheduledQuery(createScheduledQueryRequest);
final String scheduledQueryArn = createScheduledQueryResult.getArn();
System.out.println("Successfully created scheduled query : " + scheduledQueryArn);
return scheduledQueryArn;
}
catch (Exception e) {
System.out.println("Scheduled Query creation failed: " + e);
throw e;
}
}
```
------
#### [ Java v2 ]
```
public static String DATABASE_NAME = "testJavaV2DB";
public static String TABLE_NAME = "testJavaV2Table";
public static String HOSTNAME = "host-24Gju";
public static String SQ_NAME = "daily-sample";
public static String SCHEDULE_EXPRESSION = "cron(0/2 * * * ? *)";
// Find the average, p90, p95, and p99 CPU utilization for a specific EC2 host over the past 2 hours.
public static String VALID_QUERY = "SELECT region, az, hostname, BIN(time, 15s) AS binned_timestamp, " +
"ROUND(AVG(cpu_utilization), 2) AS avg_cpu_utilization, " +
"ROUND(APPROX_PERCENTILE(cpu_utilization, 0.9), 2) AS p90_cpu_utilization, " +
"ROUND(APPROX_PERCENTILE(cpu_utilization, 0.95), 2) AS p95_cpu_utilization, " +
"ROUND(APPROX_PERCENTILE(cpu_utilization, 0.99), 2) AS p99_cpu_utilization " +
"FROM " + DATABASE_NAME + "." + TABLE_NAME + " " +
"WHERE measure_name = 'metrics' " +
"AND hostname = '" + HOSTNAME + "' " +
"AND time > ago(2h) " +
"GROUP BY region, hostname, az, BIN(time, 15s) " +
"ORDER BY binned_timestamp ASC " +
"LIMIT 5";
private String createScheduledQueryHelper(String topicArn, String roleArn,
String s3ErrorReportBucketName, String query,
TargetConfiguration targetConfiguration) {
System.out.println("Creating Scheduled Query");
CreateScheduledQueryRequest createScheduledQueryRequest = CreateScheduledQueryRequest.builder()
.name(SQ_NAME)
.queryString(query)
.scheduleConfiguration(ScheduleConfiguration.builder()
.scheduleExpression(SCHEDULE_EXPRESSION)
.build())
.notificationConfiguration(NotificationConfiguration.builder()
.snsConfiguration(SnsConfiguration.builder()
.topicArn(topicArn)
.build())
.build())
.targetConfiguration(targetConfiguration)
.errorReportConfiguration(ErrorReportConfiguration.builder()
.s3Configuration(S3Configuration.builder()
.bucketName(s3ErrorReportBucketName)
.objectKeyPrefix(SCHEDULED_QUERY_EXAMPLE)
.build())
.build())
.scheduledQueryExecutionRoleArn(roleArn)
.build();
try {
final CreateScheduledQueryResponse response = queryClient.createScheduledQuery(createScheduledQueryRequest);
final String scheduledQueryArn = response.arn();
System.out.println("Successfully created scheduled query : " + scheduledQueryArn);
return scheduledQueryArn;
}
catch (Exception e) {
System.out.println("Scheduled Query creation failed: " + e);
throw e;
}
}
public String createScheduledQuery(String topicArn, String roleArn,
String databaseName, String tableName, String s3ErrorReportBucketName) {
List> sourceColToMeasureValueTypes = Arrays.asList(
Pair.of("avg_cpu_utilization", DOUBLE),
Pair.of("p90_cpu_utilization", DOUBLE),
Pair.of("p95_cpu_utilization", DOUBLE),
Pair.of("p99_cpu_utilization", DOUBLE));
TargetConfiguration targetConfiguration = TargetConfiguration.builder()
.timestreamConfiguration(TimestreamConfiguration.builder()
.databaseName(databaseName)
.tableName(tableName)
.timeColumn("binned_timestamp")
.dimensionMappings(Arrays.asList(
DimensionMapping.builder()
.name("region")
.dimensionValueType("VARCHAR")
.build(),
DimensionMapping.builder()
.name("az")
.dimensionValueType("VARCHAR")
.build(),
DimensionMapping.builder()
.name("hostname")
.dimensionValueType("VARCHAR")
.build()
))
.multiMeasureMappings(MultiMeasureMappings.builder()
.targetMultiMeasureName("multi-metrics")
.multiMeasureAttributeMappings(
sourceColToMeasureValueTypes.stream()
.map(pair -> MultiMeasureAttributeMapping.builder()
.measureValueType(pair.getValue().name())
.sourceColumn(pair.getKey())
.build())
.collect(Collectors.toList()))
.build())
.build())
.build();
return createScheduledQueryHelper(topicArn, roleArn, s3ErrorReportBucketName, VALID_QUERY, targetConfiguration);
}}
```
------
#### [ Go ]
```
SQ_ERROR_CONFIGURATION_S3_BUCKET_NAME_PREFIX = "sq-error-configuration-sample-s3-bucket-"
HOSTNAME = "host-24Gju"
SQ_NAME = "daily-sample"
SCHEDULE_EXPRESSION = "cron(0/1 * * * ? *)"
QUERY = "SELECT region, az, hostname, BIN(time, 15s) AS binned_timestamp, " +
"ROUND(AVG(cpu_utilization), 2) AS avg_cpu_utilization, " +
"ROUND(APPROX_PERCENTILE(cpu_utilization, 0.9), 2) AS p90_cpu_utilization, " +
"ROUND(APPROX_PERCENTILE(cpu_utilization, 0.95), 2) AS p95_cpu_utilization, " +
"ROUND(APPROX_PERCENTILE(cpu_utilization, 0.99), 2) AS p99_cpu_utilization " +
"FROM %s.%s " +
"WHERE measure_name = 'metrics' " +
"AND hostname = '" + HOSTNAME + "' " +
"AND time > ago(2h) " +
"GROUP BY region, hostname, az, BIN(time, 15s) " +
"ORDER BY binned_timestamp ASC " +
"LIMIT 5"
s3BucketName = utils.SQ_ERROR_CONFIGURATION_S3_BUCKET_NAME_PREFIX + generateRandomStringWithSize(5)
func generateRandomStringWithSize(size int) string {
rand.Seed(time.Now().UnixNano())
alphaNumericList := []rune("abcdefghijklmnopqrstuvwxyz0123456789")
randomPrefix := make([]rune, size)
for i := range randomPrefix {
randomPrefix[i] = alphaNumericList[rand.Intn(len(alphaNumericList))]
}
return string(randomPrefix)
}
func (timestreamBuilder TimestreamBuilder) createScheduledQuery(topicArn string, roleArn string, s3ErrorReportBucketName string,
query string, targetConfiguration timestreamquery.TargetConfiguration) (string, error) {
createScheduledQueryInput := ×treamquery.CreateScheduledQueryInput{
Name: aws.String(SQ_NAME),
QueryString: aws.String(query),
ScheduleConfiguration: ×treamquery.ScheduleConfiguration{
ScheduleExpression: aws.String(SCHEDULE_EXPRESSION),
},
NotificationConfiguration: ×treamquery.NotificationConfiguration{
SnsConfiguration: ×treamquery.SnsConfiguration{
TopicArn: aws.String(topicArn),
},
},
TargetConfiguration: &targetConfiguration,
ErrorReportConfiguration: ×treamquery.ErrorReportConfiguration{
S3Configuration: ×treamquery.S3Configuration{
BucketName: aws.String(s3ErrorReportBucketName),
},
},
ScheduledQueryExecutionRoleArn: aws.String(roleArn),
}
createScheduledQueryOutput, err := timestreamBuilder.QuerySvc.CreateScheduledQuery(createScheduledQueryInput)
if err != nil {
fmt.Printf("Error: %s", err.Error())
} else {
fmt.Println("createScheduledQueryResult is successful")
return *createScheduledQueryOutput.Arn, nil
}
return "", err
}
func (timestreamBuilder TimestreamBuilder) CreateValidScheduledQuery(topicArn string, roleArn string, s3ErrorReportBucketName string,
sqDatabaseName string, sqTableName string, databaseName string, tableName string) (string, error) {
targetConfiguration := timestreamquery.TargetConfiguration{
TimestreamConfiguration: ×treamquery.TimestreamConfiguration{
DatabaseName: aws.String(sqDatabaseName),
TableName: aws.String(sqTableName),
TimeColumn: aws.String("binned_timestamp"),
DimensionMappings: []*timestreamquery.DimensionMapping{
{
Name: aws.String("region"),
DimensionValueType: aws.String("VARCHAR"),
},
{
Name: aws.String("az"),
DimensionValueType: aws.String("VARCHAR"),
},
{
Name: aws.String("hostname"),
DimensionValueType: aws.String("VARCHAR"),
},
},
MultiMeasureMappings: ×treamquery.MultiMeasureMappings{
TargetMultiMeasureName: aws.String("multi-metrics"),
MultiMeasureAttributeMappings: []*timestreamquery.MultiMeasureAttributeMapping{
{
SourceColumn: aws.String("avg_cpu_utilization"),
MeasureValueType: aws.String(timestreamquery.MeasureValueTypeDouble),
},
{
SourceColumn: aws.String("p90_cpu_utilization"),
MeasureValueType: aws.String(timestreamquery.MeasureValueTypeDouble),
},
{
SourceColumn: aws.String("p95_cpu_utilization"),
MeasureValueType: aws.String(timestreamquery.MeasureValueTypeDouble),
},
{
SourceColumn: aws.String("p99_cpu_utilization"),
MeasureValueType: aws.String(timestreamquery.MeasureValueTypeDouble),
},
},
},
},
}
return timestreamBuilder.createScheduledQuery(topicArn, roleArn, s3ErrorReportBucketName,
fmt.Sprintf(QUERY, databaseName, tableName), targetConfiguration)
}
```
------
#### [ Python ]
```
HOSTNAME = "host-24Gju"
SQ_NAME = "daily-sample"
ERROR_BUCKET_NAME = "scheduledquerysamplerrorbucket" + ''.join([choice(ascii_lowercase) for _ in range(5)])
QUERY = \
"SELECT region, az, hostname, BIN(time, 15s) AS binned_timestamp, " \
" ROUND(AVG(cpu_utilization), 2) AS avg_cpu_utilization, " \
" ROUND(APPROX_PERCENTILE(cpu_utilization, 0.9), 2) AS p90_cpu_utilization, " \
" ROUND(APPROX_PERCENTILE(cpu_utilization, 0.95), 2) AS p95_cpu_utilization, " \
" ROUND(APPROX_PERCENTILE(cpu_utilization, 0.99), 2) AS p99_cpu_utilization " \
"FROM " + database_name + "." + table_name + " " \
"WHERE measure_name = 'metrics' " \
"AND hostname = '" + self.HOSTNAME + "' " \
"AND time > ago(2h) " \
"GROUP BY region, hostname, az, BIN(time, 15s) " \
"ORDER BY binned_timestamp ASC " \
"LIMIT 5"
def create_scheduled_query_helper(self, topic_arn, role_arn, query, target_configuration):
print("\nCreating Scheduled Query")
schedule_configuration = {
'ScheduleExpression': 'cron(0/2 * * * ? *)'
}
notification_configuration = {
'SnsConfiguration': {
'TopicArn': topic_arn
}
}
error_report_configuration = {
'S3Configuration': {
'BucketName': ERROR_BUCKET_NAME
}
}
try:
create_scheduled_query_response = \
query_client.create_scheduled_query(Name=self.SQ_NAME,
QueryString=query,
ScheduleConfiguration=schedule_configuration,
NotificationConfiguration=notification_configuration,
TargetConfiguration=target_configuration,
ScheduledQueryExecutionRoleArn=role_arn,
ErrorReportConfiguration=error_report_configuration
)
print("Successfully created scheduled query : ", create_scheduled_query_response['Arn'])
return create_scheduled_query_response['Arn']
except Exception as err:
print("Scheduled Query creation failed:", err)
raise err
def create_valid_scheduled_query(self, topic_arn, role_arn):
target_configuration = {
'TimestreamConfiguration': {
'DatabaseName': self.sq_database_name,
'TableName': self.sq_table_name,
'TimeColumn': 'binned_timestamp',
'DimensionMappings': [
{'Name': 'region', 'DimensionValueType': 'VARCHAR'},
{'Name': 'az', 'DimensionValueType': 'VARCHAR'},
{'Name': 'hostname', 'DimensionValueType': 'VARCHAR'}
],
'MultiMeasureMappings': {
'TargetMultiMeasureName': 'target_name',
'MultiMeasureAttributeMappings': [
{'SourceColumn': 'avg_cpu_utilization', 'MeasureValueType': 'DOUBLE',
'TargetMultiMeasureAttributeName': 'avg_cpu_utilization'},
{'SourceColumn': 'p90_cpu_utilization', 'MeasureValueType': 'DOUBLE',
'TargetMultiMeasureAttributeName': 'p90_cpu_utilization'},
{'SourceColumn': 'p95_cpu_utilization', 'MeasureValueType': 'DOUBLE',
'TargetMultiMeasureAttributeName': 'p95_cpu_utilization'},
{'SourceColumn': 'p99_cpu_utilization', 'MeasureValueType': 'DOUBLE',
'TargetMultiMeasureAttributeName': 'p99_cpu_utilization'},
]
}
}
}
return self.create_scheduled_query_helper(topic_arn, role_arn, QUERY, target_configuration)
```
------
#### [ Node.js ]
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/blob/mainline/sample_apps_reinvent2021/js/schedule-query-example.js)に基づいています。
```
const DATABASE_NAME = 'devops_sample_application';
const TABLE_NAME = 'host_metrics_sample_application';
const SQ_DATABASE_NAME = 'sq_result_database';
const SQ_TABLE_NAME = 'sq_result_table';
const HOSTNAME = "host-24Gju";
const SQ_NAME = "daily-sample";
const SCHEDULE_EXPRESSION = "cron(0/1 * * * ? *)";
// Find the average, p90, p95, and p99 CPU utilization for a specific EC2 host over the past 2 hours.
const VALID_QUERY = "SELECT region, az, hostname, BIN(time, 15s) AS binned_timestamp, " +
" ROUND(AVG(cpu_utilization), 2) AS avg_cpu_utilization, " +
" ROUND(APPROX_PERCENTILE(cpu_utilization, 0.9), 2) AS p90_cpu_utilization, " +
" ROUND(APPROX_PERCENTILE(cpu_utilization, 0.95), 2) AS p95_cpu_utilization, " +
" ROUND(APPROX_PERCENTILE(cpu_utilization, 0.99), 2) AS p99_cpu_utilization " +
"FROM " + DATABASE_NAME + "." + TABLE_NAME + " " +
"WHERE measure_name = 'metrics' " +
" AND hostname = '" + HOSTNAME + "' " +
" AND time > ago(2h) " +
"GROUP BY region, hostname, az, BIN(time, 15s) " +
"ORDER BY binned_timestamp ASC " +
"LIMIT 5";
async function createScheduledQuery(topicArn, roleArn, s3ErrorReportBucketName) {
console.log("Creating Valid Scheduled Query");
const DimensionMappingList = [{
'Name': 'region',
'DimensionValueType': 'VARCHAR'
},
{
'Name': 'az',
'DimensionValueType': 'VARCHAR'
},
{
'Name': 'hostname',
'DimensionValueType': 'VARCHAR'
}
];
const MultiMeasureMappings = {
TargetMultiMeasureName: "multi-metrics",
MultiMeasureAttributeMappings: [{
'SourceColumn': 'avg_cpu_utilization',
'MeasureValueType': 'DOUBLE'
},
{
'SourceColumn': 'p90_cpu_utilization',
'MeasureValueType': 'DOUBLE'
},
{
'SourceColumn': 'p95_cpu_utilization',
'MeasureValueType': 'DOUBLE'
},
{
'SourceColumn': 'p99_cpu_utilization',
'MeasureValueType': 'DOUBLE'
},
]
}
const timestreamConfiguration = {
DatabaseName: SQ_DATABASE_NAME,
TableName: SQ_TABLE_NAME,
TimeColumn: "binned_timestamp",
DimensionMappings: DimensionMappingList,
MultiMeasureMappings: MultiMeasureMappings
}
const createScheduledQueryRequest = {
Name: SQ_NAME,
QueryString: VALID_QUERY,
ScheduleConfiguration: {
ScheduleExpression: SCHEDULE_EXPRESSION
},
NotificationConfiguration: {
SnsConfiguration: {
TopicArn: topicArn
}
},
TargetConfiguration: {
TimestreamConfiguration: timestreamConfiguration
},
ScheduledQueryExecutionRoleArn: roleArn,
ErrorReportConfiguration: {
S3Configuration: {
BucketName: s3ErrorReportBucketName
}
}
};
try {
const data = await queryClient.createScheduledQuery(createScheduledQueryRequest).promise();
console.log("Successfully created scheduled query: " + data.Arn);
return data.Arn;
} catch (err) {
console.log("Scheduled Query creation failed: ", err);
throw err;
}
}
```
------
#### [ .NET ]
```
public const string Hostname = "host-24Gju";
public const string SqName = "timestream-sample";
public const string SqDatabaseName = "sq_result_database";
public const string SqTableName = "sq_result_table";
public const string ErrorConfigurationS3BucketNamePrefix = "error-configuration-sample-s3-bucket-";
public const string ScheduleExpression = "cron(0/2 * * * ? *)";
// Find the average, p90, p95, and p99 CPU utilization for a specific EC2 host over the past 2 hours.
public const string ValidQuery = "SELECT region, az, hostname, BIN(time, 15s) AS binned_timestamp, " +
"ROUND(AVG(cpu_utilization), 2) AS avg_cpu_utilization, " +
"ROUND(APPROX_PERCENTILE(cpu_utilization, 0.9), 2) AS p90_cpu_utilization, " +
"ROUND(APPROX_PERCENTILE(cpu_utilization, 0.95), 2) AS p95_cpu_utilization, " +
"ROUND(APPROX_PERCENTILE(cpu_utilization, 0.99), 2) AS p99_cpu_utilization " +
"FROM " + Constants.DATABASE_NAME + "." + Constants.TABLE_NAME + " " +
"WHERE measure_name = 'metrics' " +
"AND hostname = '" + Hostname + "' " +
"AND time > ago(2h) " +
"GROUP BY region, hostname, az, BIN(time, 15s) " +
"ORDER BY binned_timestamp ASC " +
"LIMIT 5";
private async Task CreateValidScheduledQuery(string topicArn, string roleArn,
string databaseName, string tableName, string s3ErrorReportBucketName)
{
List sourceColToMeasureValueTypes =
new List()
{
new()
{
SourceColumn = "avg_cpu_utilization",
MeasureValueType = MeasureValueType.DOUBLE.Value
},
new()
{
SourceColumn = "p90_cpu_utilization",
MeasureValueType = MeasureValueType.DOUBLE.Value
},
new()
{
SourceColumn = "p95_cpu_utilization",
MeasureValueType = MeasureValueType.DOUBLE.Value
},
new()
{
SourceColumn = "p99_cpu_utilization",
MeasureValueType = MeasureValueType.DOUBLE.Value
}
};
TargetConfiguration targetConfiguration = new TargetConfiguration()
{
TimestreamConfiguration = new TimestreamConfiguration()
{
DatabaseName = databaseName,
TableName = tableName,
TimeColumn = "binned_timestamp",
DimensionMappings = new List()
{
new()
{
Name = "region",
DimensionValueType = "VARCHAR"
},
new()
{
Name = "az",
DimensionValueType = "VARCHAR"
},
new()
{
Name = "hostname",
DimensionValueType = "VARCHAR"
}
},
MultiMeasureMappings = new MultiMeasureMappings()
{
TargetMultiMeasureName = "multi-metrics",
MultiMeasureAttributeMappings = sourceColToMeasureValueTypes
}
}
};
return await CreateScheduledQuery(topicArn, roleArn, s3ErrorReportBucketName,
ScheduledQueryConstants.ValidQuery, targetConfiguration);
}
private async Task CreateScheduledQuery(string topicArn, string roleArn,
string s3ErrorReportBucketName, string query, TargetConfiguration targetConfiguration)
{
try
{
Console.WriteLine("Creating Scheduled Query");
CreateScheduledQueryResponse response = await _amazonTimestreamQuery.CreateScheduledQueryAsync(
new CreateScheduledQueryRequest()
{
Name = ScheduledQueryConstants.SqName,
QueryString = query,
ScheduleConfiguration = new ScheduleConfiguration()
{
ScheduleExpression = ScheduledQueryConstants.ScheduleExpression
},
NotificationConfiguration = new NotificationConfiguration()
{
SnsConfiguration = new SnsConfiguration()
{
TopicArn = topicArn
}
},
TargetConfiguration = targetConfiguration,
ErrorReportConfiguration = new ErrorReportConfiguration()
{
S3Configuration = new S3Configuration()
{
BucketName = s3ErrorReportBucketName
}
},
ScheduledQueryExecutionRoleArn = roleArn
});
Console.WriteLine($"Successfully created scheduled query : {response.Arn}");
return response.Arn;
}
catch (Exception e)
{
Console.WriteLine($"Scheduled Query creation failed: {e}");
throw;
}
}
```
------
# スケジュールされたクエリを一覧表示する
次のコードスニペットを使用して、スケジュールされたクエリを一覧表示できます。
------
#### [ Java ]
```
public void listScheduledQueries() {
System.out.println("Listing Scheduled Query");
try {
String nextToken = null;
List scheduledQueries = new ArrayList<>();
do {
ListScheduledQueriesResult listScheduledQueriesResult =
queryClient.listScheduledQueries(new ListScheduledQueriesRequest()
.withNextToken(nextToken).withMaxResults(10));
List scheduledQueryList = listScheduledQueriesResult.getScheduledQueries();
printScheduledQuery(scheduledQueryList);
nextToken = listScheduledQueriesResult.getNextToken();
} while (nextToken != null);
}
catch (Exception e) {
System.out.println("List Scheduled Query failed: " + e);
throw e;
}
}
public void printScheduledQuery(List scheduledQueryList) {
for (ScheduledQuery scheduledQuery: scheduledQueryList) {
System.out.println(scheduledQuery.getArn());
}
}
```
------
#### [ Java v2 ]
```
public void listScheduledQueries() {
System.out.println("Listing Scheduled Query");
try {
String nextToken = null;
do {
ListScheduledQueriesResponse listScheduledQueriesResult =
queryClient.listScheduledQueries(ListScheduledQueriesRequest.builder()
.nextToken(nextToken).maxResults(10)
.build());
List scheduledQueryList = listScheduledQueriesResult.scheduledQueries();
printScheduledQuery(scheduledQueryList);
nextToken = listScheduledQueriesResult.nextToken();
} while (nextToken != null);
}
catch (Exception e) {
System.out.println("List Scheduled Query failed: " + e);
throw e;
}
}
public void printScheduledQuery(List scheduledQueryList) {
for (ScheduledQuery scheduledQuery: scheduledQueryList) {
System.out.println(scheduledQuery.arn());
}
}
```
------
#### [ Go ]
```
func (timestreamBuilder TimestreamBuilder) ListScheduledQueries() ([]*timestreamquery.ScheduledQuery, error) {
var nextToken *string = nil
var scheduledQueries []*timestreamquery.ScheduledQuery
for ok := true; ok; ok = nextToken != nil {
listScheduledQueriesInput := ×treamquery.ListScheduledQueriesInput{
MaxResults: aws.Int64(15),
}
if nextToken != nil {
listScheduledQueriesInput.NextToken = aws.String(*nextToken)
}
listScheduledQueriesOutput, err := timestreamBuilder.QuerySvc.ListScheduledQueries(listScheduledQueriesInput)
if err != nil {
fmt.Printf("Error: %s", err.Error())
return nil, err
}
scheduledQueries = append(scheduledQueries, listScheduledQueriesOutput.ScheduledQueries...)
nextToken = listScheduledQueriesOutput.NextToken
}
return scheduledQueries, nil
}
```
------
#### [ Python ]
```
def list_scheduled_queries(self):
print("\nListing Scheduled Queries")
try:
response = self.query_client.list_scheduled_queries(MaxResults=10)
self.print_scheduled_queries(response['ScheduledQueries'])
next_token = response.get('NextToken', None)
while next_token:
response = self.query_client.list_scheduled_queries(NextToken=next_token, MaxResults=10)
self.print_scheduled_queries(response['ScheduledQueries'])
next_token = response.get('NextToken', None)
except Exception as err:
print("List scheduled queries failed:", err)
raise err
@staticmethod
def print_scheduled_queries(scheduled_queries):
for scheduled_query in scheduled_queries:
print(scheduled_query['Arn'])
```
------
#### [ Node.js ]
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/blob/mainline/sample_apps_reinvent2021/js/schedule-query-example.js)に基づいています。
```
async function listScheduledQueries() {
console.log("Listing Scheduled Query");
try {
var nextToken = null;
do {
var params = {
MaxResults: 10,
NextToken: nextToken
}
var data = await queryClient.listScheduledQueries(params).promise();
var scheduledQueryList = data.ScheduledQueries;
printScheduledQuery(scheduledQueryList);
nextToken = data.NextToken;
}
while (nextToken != null);
} catch (err) {
console.log("List Scheduled Query failed: ", err);
throw err;
}
}
async function printScheduledQuery(scheduledQueryList) {
scheduledQueryList.forEach(element => console.log(element.Arn));
}
```
------
#### [ .NET ]
```
private async Task ListScheduledQueries()
{
try
{
Console.WriteLine("Listing Scheduled Query");
string nextToken;
do
{
ListScheduledQueriesResponse response =
await _amazonTimestreamQuery.ListScheduledQueriesAsync(new ListScheduledQueriesRequest());
foreach (var scheduledQuery in response.ScheduledQueries)
{
Console.WriteLine($"{scheduledQuery.Arn}");
}
nextToken = response.NextToken;
} while (nextToken != null);
}
catch (Exception e)
{
Console.WriteLine($"List Scheduled Query failed: {e}");
throw;
}
}
```
------
# スケジュールされたクエリを記述する
次のコードスニペットを使用して、スケジュールされたクエリを記述できます。
------
#### [ Java ]
```
public void describeScheduledQueries(String scheduledQueryArn) {
System.out.println("Describing Scheduled Query");
try {
DescribeScheduledQueryResult describeScheduledQueryResult = queryClient.describeScheduledQuery(new DescribeScheduledQueryRequest().withScheduledQueryArn(scheduledQueryArn));
System.out.println(describeScheduledQueryResult);
}
catch (ResourceNotFoundException e) {
System.out.println("Scheduled Query doesn't exist");
throw e;
}
catch (Exception e) {
System.out.println("Describe Scheduled Query failed: " + e);
throw e;
}
}
```
------
#### [ Java v2 ]
```
public void describeScheduledQueries(String scheduledQueryArn) {
System.out.println("Describing Scheduled Query");
try {
DescribeScheduledQueryResponse describeScheduledQueryResult =
queryClient.describeScheduledQuery(DescribeScheduledQueryRequest.builder()
.scheduledQueryArn(scheduledQueryArn)
.build());
System.out.println(describeScheduledQueryResult);
}
catch (ResourceNotFoundException e) {
System.out.println("Scheduled Query doesn't exist");
throw e;
}
catch (Exception e) {
System.out.println("Describe Scheduled Query failed: " + e);
throw e;
}
}
```
------
#### [ Go ]
```
func (timestreamBuilder TimestreamBuilder) DescribeScheduledQuery(scheduledQueryArn string) error {
describeScheduledQueryInput := ×treamquery.DescribeScheduledQueryInput{
ScheduledQueryArn: aws.String(scheduledQueryArn),
}
describeScheduledQueryOutput, err := timestreamBuilder.QuerySvc.DescribeScheduledQuery(describeScheduledQueryInput)
if err != nil {
if aerr, ok := err.(awserr.Error); ok {
switch aerr.Code() {
case timestreamquery.ErrCodeResourceNotFoundException:
fmt.Println(timestreamquery.ErrCodeResourceNotFoundException, aerr.Error())
default:
fmt.Printf("Error: %s", err.Error())
}
} else {
fmt.Printf("Error: %s", aerr.Error())
}
return err
} else {
fmt.Println("DescribeScheduledQuery is successful, below is the output:")
fmt.Println(describeScheduledQueryOutput.ScheduledQuery)
return nil
}
}
```
------
#### [ Python ]
```
def describe_scheduled_query(self, scheduled_query_arn):
print("\nDescribing Scheduled Query")
try:
response = self.query_client.describe_scheduled_query(ScheduledQueryArn=scheduled_query_arn)
if 'ScheduledQuery' in response:
response = response['ScheduledQuery']
for key in response:
print("{} :{}".format(key, response[key]))
except self.query_client.exceptions.ResourceNotFoundException as err:
print("Scheduled Query doesn't exist")
raise err
except Exception as err:
print("Scheduled Query describe failed:", err)
raise err
```
------
#### [ Node.js ]
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/blob/mainline/sample_apps_reinvent2021/js/schedule-query-example.js)に基づいています。
```
async function describeScheduledQuery(scheduledQueryArn) {
console.log("Describing Scheduled Query");
var params = {
ScheduledQueryArn: scheduledQueryArn
}
try {
const data = await queryClient.describeScheduledQuery(params).promise();
console.log(data.ScheduledQuery);
} catch (err) {
console.log("Describe Scheduled Query failed: ", err);
throw err;
}
}
```
------
#### [ .NET ]
```
private async Task DescribeScheduledQuery(string scheduledQueryArn)
{
try
{
Console.WriteLine("Describing Scheduled Query");
DescribeScheduledQueryResponse response = await _amazonTimestreamQuery.DescribeScheduledQueryAsync(
new DescribeScheduledQueryRequest()
{
ScheduledQueryArn = scheduledQueryArn
});
Console.WriteLine($"{JsonConvert.SerializeObject(response.ScheduledQuery)}");
}
catch (ResourceNotFoundException e)
{
Console.WriteLine($"Scheduled Query doesn't exist: {e}");
throw;
}
catch (Exception e)
{
Console.WriteLine($"Describe Scheduled Query failed: {e}");
throw;
}
}
```
------
# スケジュールされたクエリを実行する
次のコードスニペットを使用して、スケジュールされたクエリを実行できます。
------
#### [ Java ]
```
public void executeScheduledQueries(String scheduledQueryArn, Date invocationTime) {
System.out.println("Executing Scheduled Query");
try {
ExecuteScheduledQueryResult executeScheduledQueryResult = queryClient.executeScheduledQuery(new ExecuteScheduledQueryRequest()
.withScheduledQueryArn(scheduledQueryArn)
.withInvocationTime(invocationTime)
);
}
catch (ResourceNotFoundException e) {
System.out.println("Scheduled Query doesn't exist");
throw e;
}
catch (Exception e) {
System.out.println("Execution Scheduled Query failed: " + e);
throw e;
}
}
```
------
#### [ Java v2 ]
```
public void executeScheduledQuery(String scheduledQueryArn) {
System.out.println("Executing Scheduled Query");
try {
ExecuteScheduledQueryResponse executeScheduledQueryResult = queryClient.executeScheduledQuery(ExecuteScheduledQueryRequest.builder()
.scheduledQueryArn(scheduledQueryArn)
.invocationTime(Instant.now())
.build()
);
System.out.println("Execute ScheduledQuery response code: " + executeScheduledQueryResult.sdkHttpResponse().statusCode());
}
catch (ResourceNotFoundException e) {
System.out.println("Scheduled Query doesn't exist");
throw e;
}
catch (Exception e) {
System.out.println("Execution Scheduled Query failed: " + e);
throw e;
}
}
```
------
#### [ Go ]
```
func (timestreamBuilder TimestreamBuilder) ExecuteScheduledQuery(scheduledQueryArn string, invocationTime time.Time) error {
executeScheduledQueryInput := ×treamquery.ExecuteScheduledQueryInput{
ScheduledQueryArn: aws.String(scheduledQueryArn),
InvocationTime: aws.Time(invocationTime),
}
executeScheduledQueryOutput, err := timestreamBuilder.QuerySvc.ExecuteScheduledQuery(executeScheduledQueryInput)
if err != nil {
if aerr, ok := err.(awserr.Error); ok {
switch aerr.Code() {
case timestreamquery.ErrCodeResourceNotFoundException:
fmt.Println(timestreamquery.ErrCodeResourceNotFoundException, aerr.Error())
default:
fmt.Printf("Error: %s", aerr.Error())
}
} else {
fmt.Printf("Error: %s", err.Error())
}
return err
} else {
fmt.Println("ExecuteScheduledQuery is successful, below is the output:")
fmt.Println(executeScheduledQueryOutput.GoString())
return nil
}
}
```
------
#### [ Python ]
```
def execute_scheduled_query(self, scheduled_query_arn, invocation_time):
print("\nExecuting Scheduled Query")
try:
self.query_client.execute_scheduled_query(ScheduledQueryArn=scheduled_query_arn, InvocationTime=invocation_time)
print("Successfully started executing scheduled query")
except self.query_client.exceptions.ResourceNotFoundException as err:
print("Scheduled Query doesn't exist")
raise err
except Exception as err:
print("Scheduled Query execution failed:", err)
raise err
```
------
#### [ Node.js ]
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/blob/mainline/sample_apps_reinvent2021/js/schedule-query-example.js)に基づいています。
```
async function executeScheduledQuery(scheduledQueryArn, invocationTime) {
console.log("Executing Scheduled Query");
var params = {
ScheduledQueryArn: scheduledQueryArn,
InvocationTime: invocationTime
}
try {
await queryClient.executeScheduledQuery(params).promise();
} catch (err) {
console.log("Execute Scheduled Query failed: ", err);
throw err;
}
}
```
------
#### [ .NET ]
```
private async Task ExecuteScheduledQuery(string scheduledQueryArn, DateTime invocationTime)
{
try
{
Console.WriteLine("Running Scheduled Query");
await _amazonTimestreamQuery.ExecuteScheduledQueryAsync(new ExecuteScheduledQueryRequest()
{
ScheduledQueryArn = scheduledQueryArn,
InvocationTime = invocationTime
});
Console.WriteLine("Successfully started manual run of scheduled query");
}
catch (ResourceNotFoundException e)
{
Console.WriteLine($"Scheduled Query doesn't exist: {e}");
throw;
}
catch (Exception e)
{
Console.WriteLine($"Execute Scheduled Query failed: {e}");
throw;
}
}
```
------
# スケジュールされたクエリを更新する
次のコードスニペットを使用して、スケジュールされたクエリを更新できます。
------
#### [ Java ]
```
public void updateScheduledQueries(String scheduledQueryArn) {
System.out.println("Updating Scheduled Query");
try {
queryClient.updateScheduledQuery(new UpdateScheduledQueryRequest()
.withScheduledQueryArn(scheduledQueryArn)
.withState(ScheduledQueryState.DISABLED));
System.out.println("Successfully update scheduled query state");
}
catch (ResourceNotFoundException e) {
System.out.println("Scheduled Query doesn't exist");
throw e;
}
catch (Exception e) {
System.out.println("Execution Scheduled Query failed: " + e);
throw e;
}
}
```
------
#### [ Java v2 ]
```
public void updateScheduledQuery(String scheduledQueryArn, ScheduledQueryState state) {
System.out.println("Updating Scheduled Query");
try {
queryClient.updateScheduledQuery(UpdateScheduledQueryRequest.builder()
.scheduledQueryArn(scheduledQueryArn)
.state(state)
.build());
System.out.println("Successfully update scheduled query state");
}
catch (ResourceNotFoundException e) {
System.out.println("Scheduled Query doesn't exist");
throw e;
}
catch (Exception e) {
System.out.println("Execution Scheduled Query failed: " + e);
throw e;
}
}
```
------
#### [ Go ]
```
func (timestreamBuilder TimestreamBuilder) UpdateScheduledQuery(scheduledQueryArn string) error {
updateScheduledQueryInput := ×treamquery.UpdateScheduledQueryInput{
ScheduledQueryArn: aws.String(scheduledQueryArn),
State: aws.String(timestreamquery.ScheduledQueryStateDisabled),
}
_, err := timestreamBuilder.QuerySvc.UpdateScheduledQuery(updateScheduledQueryInput)
if err != nil {
if aerr, ok := err.(awserr.Error); ok {
switch aerr.Code() {
case timestreamquery.ErrCodeResourceNotFoundException:
fmt.Println(timestreamquery.ErrCodeResourceNotFoundException, aerr.Error())
default:
fmt.Printf("Error: %s", aerr.Error())
}
} else {
fmt.Printf("Error: %s", err.Error())
}
return err
} else {
fmt.Println("UpdateScheduledQuery is successful")
return nil
}
}
```
------
#### [ Python ]
```
def update_scheduled_query(self, scheduled_query_arn, state):
print("\nUpdating Scheduled Query")
try:
self.query_client.update_scheduled_query(ScheduledQueryArn=scheduled_query_arn,
State=state)
print("Successfully update scheduled query state to", state)
except self.query_client.exceptions.ResourceNotFoundException as err:
print("Scheduled Query doesn't exist")
raise err
except Exception as err:
print("Scheduled Query deletion failed:", err)
raise err
```
------
#### [ Node.js ]
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/blob/mainline/sample_apps_reinvent2021/js/schedule-query-example.js)に基づいています。
```
async function updateScheduledQueries(scheduledQueryArn) {
console.log("Updating Scheduled Query");
var params = {
ScheduledQueryArn: scheduledQueryArn,
State: "DISABLED"
}
try {
await queryClient.updateScheduledQuery(params).promise();
console.log("Successfully update scheduled query state");
} catch (err) {
console.log("Update Scheduled Query failed: ", err);
throw err;
}
}
```
------
#### [ .NET ]
```
private async Task UpdateScheduledQuery(string scheduledQueryArn, ScheduledQueryState state)
{
try
{
Console.WriteLine("Updating Scheduled Query");
await _amazonTimestreamQuery.UpdateScheduledQueryAsync(new UpdateScheduledQueryRequest()
{
ScheduledQueryArn = scheduledQueryArn,
State = state
});
Console.WriteLine("Successfully update scheduled query state");
}
catch (ResourceNotFoundException e)
{
Console.WriteLine($"Scheduled Query doesn't exist: {e}");
throw;
}
catch (Exception e)
{
Console.WriteLine($"Update Scheduled Query failed: {e}");
throw;
}
}
```
------
# スケジュールされたクエリを削除する
次のコードスニペットを使用して、スケジュールされたクエリを削除できます。
------
#### [ Java ]
```
public void deleteScheduledQuery(String scheduledQueryArn) {
System.out.println("Deleting Scheduled Query");
try {
queryClient.deleteScheduledQuery(new DeleteScheduledQueryRequest().withScheduledQueryArn(scheduledQueryArn));
System.out.println("Successfully deleted scheduled query");
}
catch (Exception e) {
System.out.println("Scheduled Query deletion failed: " + e);
}
}
```
------
#### [ Java v2 ]
```
public void deleteScheduledQuery(String scheduledQueryArn) {
System.out.println("Deleting Scheduled Query");
try {
queryClient.deleteScheduledQuery(DeleteScheduledQueryRequest.builder()
.scheduledQueryArn(scheduledQueryArn).build());
System.out.println("Successfully deleted scheduled query");
}
catch (Exception e) {
System.out.println("Scheduled Query deletion failed: " + e);
}
}
```
------
#### [ Go ]
```
func (timestreamBuilder TimestreamBuilder) DeleteScheduledQuery(scheduledQueryArn string) error {
deleteScheduledQueryInput := ×treamquery.DeleteScheduledQueryInput{
ScheduledQueryArn: aws.String(scheduledQueryArn),
}
_, err := timestreamBuilder.QuerySvc.DeleteScheduledQuery(deleteScheduledQueryInput)
if err != nil {
fmt.Println("Error:")
if aerr, ok := err.(awserr.Error); ok {
switch aerr.Code() {
case timestreamquery.ErrCodeResourceNotFoundException:
fmt.Println(timestreamquery.ErrCodeResourceNotFoundException, aerr.Error())
default:
fmt.Printf("Error: %s", aerr.Error())
}
} else {
fmt.Printf("Error: %s", err.Error())
}
return err
} else {
fmt.Println("DeleteScheduledQuery is successful")
return nil
}
}
```
------
#### [ Python ]
```
def delete_scheduled_query(self, scheduled_query_arn):
print("\nDeleting Scheduled Query")
try:
self.query_client.delete_scheduled_query(ScheduledQueryArn=scheduled_query_arn)
print("Successfully deleted scheduled query :", scheduled_query_arn)
except Exception as err:
print("Scheduled Query deletion failed:", err)
raise err
```
------
#### [ Node.js ]
以下のコード例では、AWS SDK for JavaScript V2 スタイルを使用しています。これは、[GitHub の Node.js サンプル Amazon Timestream for LiveAnalytics アプリケーション](https://github.com/awslabs/amazon-timestream-tools/blob/mainline/sample_apps_reinvent2021/js/schedule-query-example.js)に基づいています。
```
async function deleteScheduleQuery(scheduledQueryArn) {
console.log("Deleting Scheduled Query");
const params = {
ScheduledQueryArn: scheduledQueryArn
}
try {
await queryClient.deleteScheduledQuery(params).promise();
console.log("Successfully deleted scheduled query");
} catch (err) {
console.log("Scheduled Query deletion failed: ", err);
}
}
```
------
#### [ .NET ]
```
private async Task DeleteScheduledQuery(string scheduledQueryArn)
{
try
{
Console.WriteLine("Deleting Scheduled Query");
await _amazonTimestreamQuery.DeleteScheduledQueryAsync(new DeleteScheduledQueryRequest()
{
ScheduledQueryArn = scheduledQueryArn
});
Console.WriteLine($"Successfully deleted scheduled query : {scheduledQueryArn}");
}
catch (Exception e)
{
Console.WriteLine($"Scheduled Query deletion failed: {e}");
throw;
}
}
```
------