For similar capabilities to Amazon Timestream for LiveAnalytics, consider Amazon Timestream for InfluxDB. It offers simplified data ingestion and single-digit millisecond query response times for real-time analytics. Learn more [here](https://docs.aws.amazon.com//timestream/latest/developerguide/timestream-for-influxdb.html).
# Code samples
You can access Amazon Timestream using the AWS SDKs. Timestream supports two SDKs per language; namely, the Write SDK and the Query SDK. The Write SDK is used to perform CRUD operations and to insert your time series data into Timestream. The Query SDK is used to query your existing time series data stored in Timestream. Select a topic from the list below for more details, including code samples for each of the supported SDKs.
**Topics**
+ [
# Write SDK client
](code-samples.write-client.md)
+ [
# Query SDK client
](code-samples.query-client.md)
+ [
# Create database
](code-samples.create-db.md)
+ [
# Describe database
](code-samples.describe-db.md)
+ [
# Update database
](code-samples.update-db.md)
+ [
# Delete database
](code-samples.delete-db.md)
+ [
# List databases
](code-samples.list-db.md)
+ [
# Create table
](code-samples.create-table.md)
+ [
# Describe table
](code-samples.describe-table.md)
+ [
# Update table
](code-samples.update-table.md)
+ [
# Delete table
](code-samples.delete-table.md)
+ [
# List tables
](code-samples.list-table.md)
+ [
# Write data (inserts and upserts)
](code-samples.write.md)
+ [
# Run query
](code-samples.run-query.md)
+ [
# Run UNLOAD query
](code-samples.run-query-unload.md)
+ [
# Cancel query
](code-samples.cancel-query.md)
+ [
# Create batch load task
](code-samples.create-batch-load.md)
+ [
# Describe batch load task
](code-samples.describe-batch-load.md)
+ [
# List batch load tasks
](code-samples.list-batch-load-tasks.md)
+ [
# Resume batch load task
](code-samples.resume-batch-load-task.md)
+ [
# Create scheduled query
](code-samples.create-scheduledquery.md)
+ [
# List scheduled query
](code-samples.list-scheduledquery.md)
+ [
# Describe scheduled query
](code-samples.describe-scheduledquery.md)
+ [
# Execute scheduled query
](code-samples.execute-scheduledquery.md)
+ [
# Update scheduled query
](code-samples.update-scheduledquery.md)
+ [
# Delete scheduled query
](code-samples.delete-scheduledquery.md)
# Write SDK client
You can use the following code snippets to create a Timestream client for the Write SDK. The Write SDK is used to perform CRUD operations and to insert your time series data into Timestream.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses AWS SDK for JavaScript v3. For more information about how to install the client and usage, see [Timestream Write Client - AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html).
An additional command import is shown here. The `CreateDatabaseCommand` import is not required to create the client.
```
import { TimestreamWriteClient, CreateDatabaseCommand } from "@aws-sdk/client-timestream-write";
const writeClient = new TimestreamWriteClient({ region: "us-east-1" });
```
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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);
```
------
We recommend you use the following configuration.
+ Set the SDK retry count to `10`.
+ Use `SDK DEFAULT_BACKOFF_STRATEGY`.
+ Set `RequestTimeout` to `20` seconds.
+ Set the max connections to `5000` or higher.
# Query SDK client
You can use the following code snippets to create a Timestream client for the Query SDK. The Query SDK is used to query your existing time series data stored in Timestream.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses AWS SDK for JavaScript v3. For more information about how to install the client and usage, see [Timestream Query Client - ,AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-query/index.html).
An additional command import is shown here. The `QueryCommand` import is not required to create the client.
```
import { TimestreamQueryClient, QueryCommand } from "@aws-sdk/client-timestream-query";
const queryClient = new TimestreamQueryClient({ region: "us-east-1" });
```
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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);
```
------
# Create database
You can use the following code snippets to create a database.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses AWS SDK for JavaScript v3. For more information about how to install the client and usage, see [Timestream Write Client - AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html).
Also see [Class CreateDatabaseCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/createdatabasecommand.html) and [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);
}
}
```
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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());
}
}
```
------
# Describe database
You can use the following code snippets to get information about the attributes of your newly created database.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses AWS SDK for JavaScript v3. For more information about how to install the client and usage, see [Timestream Write Client - AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html).
Also see [Class DescribeDatabaseCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/describedatabasecommand.html) and [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;
}
}
```
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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());
}
}
```
------
# Update database
You can use the following code snippets to update your databases.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses AWS SDK for JavaScript v3. For more information about how to install the client and usage, see [Timestream Write Client - AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html).
Also see [Class UpdateDatabaseCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/updatedatabasecommand.html) and [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);
}
}
```
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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}");
}
```
------
# Delete database
You can use the following code snippet to delete a database.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses AWS SDK for JavaScript v3. For more information about how to install the client and usage, see [Timestream Write Client - AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html).
Also see [Class DeleteDatabaseCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/deletedatabasecommand.html) and [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;
}
}
```
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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());
}
}
```
------
# List databases
You can use the following code snippets to list your databases.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses AWS SDK for JavaScript v3. For more information about how to install the client and usage, see [Timestream Write Client - AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html).
Also see [Class ListDatabasesCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/listdatabasescommand.html) and [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);
}
}
```
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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());
}
}
```
------
# Create table
**Topics**
+ [
## Memory store writes
](#code-samples.create-table-memorystore)
+ [
## Magnetic store writes
](#code-samples.create-table-magneticstore)
## Memory store writes
You can use the following code snippet to create a table that has magnetic store writes disabled, as a result you can only write data into your memory store retention window.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses AWS SDK for JavaScript v3. For more information about how to install the client and usage, see [Timestream Write Client - AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html).
Also see [Class CreateTableCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/createtablecommand.html) and [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;
}
}
```
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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());
}
}
```
------
## Magnetic store writes
You can use the following code snippet to create a table with magnetic store writes enabled. With magnetic store writes you can write data into both your memory store retention window and magnetic store retention window.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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());
}
}
```
------
# Describe table
You can use the following code snippets to get information about the attributes of your table.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses AWS SDK for JavaScript v3. For more information about how to install the client and usage, see [Timestream Write Client - AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html).
Also see [Class DescribeTableCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/describedatabasecommand.html) and [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;
}
}
```
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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());
}
}
```
------
# Update table
You can use the following code snippets to update a table.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses AWS SDK for JavaScript v3. For more information about how to install the client and usage, see [Timestream Write Client - AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html).
Also see [Class UpdateTableCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/updatetablecommand.html) and [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);
}
```
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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());
}
}
```
------
# Delete table
You can use the following code snippets to delete a table.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses AWS SDK for JavaScript v3. For more information about how to install the client and usage, see [Timestream Write Client - AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html).
Also see [Class DeleteTableCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/deletetablecommand.html) and [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;
}
}
```
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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());
}
}
```
------
# List tables
You can use the following code snippets to list tables.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses AWS SDK for JavaScript v3. For more information about how to install the client and usage, see [Timestream Write Client - AWS SDK for JavaScript v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html).
Also see [Class ListTablesCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/listtablescommand.html) and [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);
}
}
```
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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}");
}
```
------
# Write data (inserts and upserts)
**Topics**
+ [
## Writing batches of records
](#code-samples.write.write-batches)
+ [
## Writing batches of records with common attributes
](#code-samples.write.write-batches-common-attrs)
+ [
## Upserting records
](#code-samples.write.upserts)
+ [
## Multi-measure attribute example
](#code-samples.write.data.multivalue)
+ [
## Handling write failures
](#code-samples.write.rejectedRecordException)
## Writing batches of records
You can use the following code snippets to write data into an Amazon Timestream table. Writing data in batches helps to optimize the cost of writes. See [Calculating the number of writes](metering-and-pricing.writes.md#metering-and-pricing.writes.write-size-multiple-events) for more information.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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());
}
}
```
------
## Writing batches of records with common attributes
If your time series data has measures and/or dimensions that are common across many data points, you can also use the following optimized version of the writeRecords API to insert data into Timestream for LiveAnalytics. Using common attributes with batching can further optimize the cost of writes as described in [Calculating the number of writes](metering-and-pricing.writes.md#metering-and-pricing.writes.write-size-multiple-events).
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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());
}
}
```
------
## Upserting records
While the default writes in Amazon Timestream follow the *first writer wins* semantics, where data is stored as append only and duplicate records are rejected, there are applications that require the ability to write data into Amazon Timestream using the *last writer wins* semantics, where the record with the highest version is stored in the system. There are also applications that require the ability to update existing records. To address these scenarios, Amazon Timestream provides the ability to *upsert* data. Upsert is an operation that inserts a record in to the system when the record does not exist or updates the record, when one exists.
You can upsert records by including the `Version` in record definition while sending a `WriteRecords` request. Amazon Timestream will store the record with the record with highest `Version`. The code sample below shows how you can upsert data:
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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());
}
}
```
------
## Multi-measure attribute example
This example illustrates writing multi-mearure attributes. [Multi-measure attributes](data-modeling.md#data-modeling-multiVsinglerecords) are useful when a device or an application you are tracking emits multiple metrics or events at the same timestamp..
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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());
}
}
}
}
```
------
## Handling write failures
Writes in Amazon Timestream can fail for one or more of the following reasons:
+ There are records with timestamps that lie outside the retention duration of the memory store.
+ There are records containing dimensions and/or measures that exceed the Timestream defined limits.
+ Amazon Timestream has detected duplicate records. Records are marked as duplicate, when there are multiple records with the same dimensions, timestamps, and measure names but:
+ Measure values are different.
+ Version is not present in the request or the value of version in the new record is equal to or lower than the existing value. If Amazon Timestream rejects data for this reason, the `ExistingVersion` field in the `RejectedRecords` will contain the record's current version as stored in Amazon Timestream. To force an update, you can resend the request with a version for the record set to a value greater than the `ExistingVersion`.
For more information about errors and rejected records, see [Errors](https://docs.aws.amazon.com/timestream/latest/developerguide/API_WriteRecords.html#API_WriteRecords_Errors) and [RejectedRecord](https://docs.aws.amazon.com/timestream/latest/developerguide/API_RejectedRecord.html).
If your application receives a `RejectedRecordsException` when attempting to write records to Timestream, you can parse the rejected records to learn more about the write failures as shown below.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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());
}
```
------
# Run query
**Topics**
+ [
## Paginating results
](#code-samples.run-query.pagination)
+ [
## Parsing result sets
](#code-samples.run-query.parsing)
+ [
## Accessing the query status
](#code-samples.run-query.query-status)
## Paginating results
When you run a query, Timestream returns the result set in a paginated manner to optimize the responsiveness of your applications. The code snippet below shows how you can paginate through the result set. You must loop through all the result set pages until you encounter a null value. Pagination tokens expire 3 hours after being issued by Timestream for LiveAnalytics.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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());
}
}
```
------
## Parsing result sets
You can use the following code snippets to extract data from the result set. Query results are accessible for up to 24 hours after a query completes.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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)}]";
}
```
------
## Accessing the query status
You can access the query status through `QueryResponse`, which contains information about progress of a query, the bytes scanned by a query and the bytes metered by a query. The `bytesMetered` and `bytesScanned` values are cumulative and continuously updated while paging query results. You can use this information to understand the bytes scanned by an individual query and also use it to make certain decisions. For example, assuming that the query price is \$10.01 per GB scanned, you may want to cancel queries that exceed \$125 per query, or `X` GB. The code snippet below shows how this can be done.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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());
}
}
```
------
For additional details on how to cancel a query, see [Cancel query](code-samples.cancel-query.md).
# Run UNLOAD query
The following code examples call an UNLOAD query. For information about `UNLOAD`, see [Using UNLOAD to export query results to S3 from Timestream for LiveAnalytics](export-unload.md). For examples of `UNLOAD` queries, see [Example use case for UNLOAD from Timestream for LiveAnalytics](export-unload-example-use-case.md).
**Topics**
+ [
## Build and run an UNLOAD query
](#code-samples.run-query-unload-build-and-run)
+ [
## Parse UNLOAD response, and get row count, manifest link, and metadata link
](#code-samples.run-query-unload-parse-response)
+ [
## Read and parse manifest content
](#code-samples.run-query-unload-parse-manifest)
+ [
## Read and parse metadata content
](#code-samples.run-query-unload-parse-metadata)
## Build and run an UNLOAD query
------
#### [ 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
}
}
```
------
## Parse UNLOAD response, and get row count, manifest link, and metadata link
------
#### [ 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
}
```
------
## Read and parse manifest content
------
#### [ 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};
}
```
------
## Read and parse metadata content
------
#### [ 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};
}
```
------
# Cancel query
You can use the following code snippets to cancel a query.
**Note**
These code snippets are based on full sample applications on [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps). For more information about how to get started with the sample applications, see [Sample application](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 ]
The following snippet uses the AWS SDK for JavaScript V2 style. It is based on the sample application at [Node.js sample Amazon Timestream for LiveAnalytics application on GitHub](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);
}
}
```
------
# Create batch load task
You can use the following code snippets to create batch load tasks.
------
#### [ 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 =