Für ähnliche Funktionen wie Amazon Timestream für sollten Sie Amazon Timestream for LiveAnalytics InfluxDB in Betracht ziehen. Es bietet eine vereinfachte Datenaufnahme und Antwortzeiten im einstelligen Millisekundenbereich für Analysen in Echtzeit. [Erfahren](https://docs.aws.amazon.com//timestream/latest/developerguide/timestream-for-influxdb.html) Sie hier mehr.
Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.
# Codebeispiele
Sie können mit dem auf Amazon Timestream zugreifen. AWS SDKs Timestream unterstützt zwei SDKs pro Sprache, nämlich das Write SDK und das Query SDK. Das Write SDK wird verwendet, um CRUD-Operationen durchzuführen und Ihre Zeitreihendaten in Timestream einzufügen. Das Query SDK wird verwendet, um Ihre vorhandenen Zeitreihendaten abzufragen, die in Timestream gespeichert sind. Wählen Sie ein Thema aus der folgenden Liste aus, um weitere Informationen zu erhalten, einschließlich Codebeispielen für jedes der unterstützten SDKs Themen.
**Topics**
+ [
# SDK-Client schreiben
](code-samples.write-client.md)
+ [
# SDK-Client abfragen
](code-samples.query-client.md)
+ [
# Datenbank erstellen
](code-samples.create-db.md)
+ [
# Beschreiben Sie die Datenbank
](code-samples.describe-db.md)
+ [
# Datenbank aktualisieren
](code-samples.update-db.md)
+ [
# Datenbank löschen
](code-samples.delete-db.md)
+ [
# Datenbanken auflisten
](code-samples.list-db.md)
+ [
# Create table
](code-samples.create-table.md)
+ [
# Tabelle beschreiben
](code-samples.describe-table.md)
+ [
# Tabelle aktualisieren
](code-samples.update-table.md)
+ [
# Tabelle löschen
](code-samples.delete-table.md)
+ [
# Auflisten von Tabellen
](code-samples.list-table.md)
+ [
# Daten schreiben (Einfügungen und Upserts)
](code-samples.write.md)
+ [
# Abfrage ausführen
](code-samples.run-query.md)
+ [
# UNLOAD-Abfrage ausführen
](code-samples.run-query-unload.md)
+ [
# Anfrage abbrechen
](code-samples.cancel-query.md)
+ [
# Batch-Load-Aufgabe erstellen
](code-samples.create-batch-load.md)
+ [
# Beschreiben Sie die Batch-Load-Aufgabe
](code-samples.describe-batch-load.md)
+ [
# Batch-Load-Aufgaben auflisten
](code-samples.list-batch-load-tasks.md)
+ [
# Batch-Load-Aufgabe fortsetzen
](code-samples.resume-batch-load-task.md)
+ [
# Geplante Abfrage erstellen
](code-samples.create-scheduledquery.md)
+ [
# Geplante Abfragen auflisten
](code-samples.list-scheduledquery.md)
+ [
# Geplante Abfrage beschreiben
](code-samples.describe-scheduledquery.md)
+ [
# Geplante Abfrage ausführen
](code-samples.execute-scheduledquery.md)
+ [
# Geplante Abfrage aktualisieren
](code-samples.update-scheduledquery.md)
+ [
# Geplante Abfrage löschen
](code-samples.delete-scheduledquery.md)
# SDK-Client schreiben
Sie können die folgenden Codefragmente verwenden, um einen Timestream-Client für das Write SDK zu erstellen. Das Write SDK wird verwendet, um CRUD-Operationen durchzuführen und Ihre Zeitreihendaten in Timestream einzufügen.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Im folgenden Codeausschnitt wird AWS SDK für JavaScript Version 3 verwendet. Weitere Informationen zur Installation und Verwendung des Clients finden Sie unter [Timestream Write Client — AWS SDK for v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html). JavaScript
Ein zusätzlicher Befehlsimport wird hier gezeigt. Der `CreateDatabaseCommand` Import ist nicht erforderlich, um den Client zu erstellen.
```
import { TimestreamWriteClient, CreateDatabaseCommand } from "@aws-sdk/client-timestream-write";
const writeClient = new TimestreamWriteClient({ region: "us-east-1" });
```
Das folgende Snippet verwendet das AWS SDK für JavaScript V2-Stil. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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);
```
------
Wir empfehlen Ihnen, die folgende Konfiguration zu verwenden.
+ Stellen Sie die Anzahl der SDK-Wiederholungsversuche auf ein`10`.
+ Verwenden Sie `SDK DEFAULT_BACKOFF_STRATEGY`.
+ Stellen Sie den `RequestTimeout` Wert auf `20` Sekunden ein.
+ Stellen Sie die maximale Anzahl an Verbindungen auf `5000` oder höher ein.
# SDK-Client abfragen
Sie können die folgenden Codefragmente verwenden, um einen Timestream-Client für das Query SDK zu erstellen. Das Query-SDK wird verwendet, um Ihre vorhandenen Zeitreihendaten abzufragen, die in Timestream gespeichert sind.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Im folgenden Codeausschnitt wird AWS SDK für JavaScript Version 3 verwendet. Weitere Informationen zur Installation und Verwendung des Clients finden Sie unter [Timestream Query Client -,AWS SDK](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-query/index.html) für v3. JavaScript
Ein zusätzlicher Befehlsimport wird hier gezeigt. Der `QueryCommand` Import ist nicht erforderlich, um den Client zu erstellen.
```
import { TimestreamQueryClient, QueryCommand } from "@aws-sdk/client-timestream-query";
const queryClient = new TimestreamQueryClient({ region: "us-east-1" });
```
Das folgende Snippet verwendet das AWS SDK für JavaScript V2-Stil. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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);
```
------
# Datenbank erstellen
Sie können die folgenden Codefragmente verwenden, um eine Datenbank zu erstellen.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Im folgenden Codeausschnitt wird AWS SDK für JavaScript Version 3 verwendet. Weitere Informationen zur Installation und Verwendung des Clients finden Sie unter [Timestream Write Client — AWS SDK for v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html). JavaScript
Siehe auch [Klasse CreateDatabaseCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/createdatabasecommand.html) und [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);
}
}
```
Das folgende Snippet verwendet das AWS SDK für den JavaScript V2-Stil. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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());
}
}
```
------
# Beschreiben Sie die Datenbank
Sie können die folgenden Codefragmente verwenden, um Informationen über die Attribute Ihrer neu erstellten Datenbank zu erhalten.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Im folgenden Codeausschnitt wird AWS SDK für JavaScript Version 3 verwendet. Weitere Informationen zur Installation und Verwendung des Clients finden Sie unter [Timestream Write Client — AWS SDK for v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html). JavaScript
Siehe auch [Klasse DescribeDatabaseCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/describedatabasecommand.html) und [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;
}
}
```
Das folgende Snippet verwendet das AWS SDK für den JavaScript V2-Stil. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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());
}
}
```
------
# Datenbank aktualisieren
Sie können die folgenden Codefragmente verwenden, um Ihre Datenbanken zu aktualisieren.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Der folgende Codeausschnitt verwendet AWS SDK für Version 3. JavaScript Weitere Informationen zur Installation und Verwendung des Clients finden Sie unter [Timestream Write Client — AWS SDK for v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html). JavaScript
Siehe auch [Klasse UpdateDatabaseCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/updatedatabasecommand.html) und [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);
}
}
```
Das folgende Snippet verwendet das AWS SDK für den JavaScript V2-Stil. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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}");
}
```
------
# Datenbank löschen
Sie können den folgenden Codeausschnitt verwenden, um eine Datenbank zu löschen.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Im folgenden Codeausschnitt wird AWS SDK für JavaScript Version 3 verwendet. Weitere Informationen zur Installation und Verwendung des Clients finden Sie unter [Timestream Write Client — AWS SDK for v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html). JavaScript
Siehe auch [Klasse DeleteDatabaseCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/deletedatabasecommand.html) und [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;
}
}
```
Das folgende Snippet verwendet das AWS SDK für den JavaScript V2-Stil. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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());
}
}
```
------
# Datenbanken auflisten
Sie können die folgenden Codefragmente verwenden, um Ihre Datenbanken aufzulisten.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Der folgende Codeausschnitt verwendet AWS SDK für Version 3. JavaScript Weitere Informationen zur Installation und Verwendung des Clients finden Sie unter [Timestream Write Client — AWS SDK for v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html). JavaScript
Siehe auch [Klasse ListDatabasesCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/listdatabasescommand.html) und [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);
}
}
```
Das folgende Snippet verwendet das AWS SDK für den JavaScript V2-Stil. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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**
+ [
## Speichert Schreibvorgänge
](#code-samples.create-table-memorystore)
+ [
## Magnetic Store schreibt
](#code-samples.create-table-magneticstore)
## Speichert Schreibvorgänge
Sie können den folgenden Codeausschnitt verwenden, um eine Tabelle zu erstellen, in der Magnetspeicher-Schreibvorgänge deaktiviert sind. Daher können Sie nur Daten in das Aufbewahrungsfenster für den Speicherspeicher schreiben.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Im folgenden Codeausschnitt wird AWS SDK für JavaScript Version 3 verwendet. Weitere Informationen zur Installation und Verwendung des Clients finden Sie unter [Timestream Write Client — AWS SDK for v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html). JavaScript
Siehe auch [Klasse CreateTableCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/createtablecommand.html) und [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;
}
}
```
Das folgende Snippet verwendet das AWS SDK für den JavaScript V2-Stil. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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 schreibt
Sie können den folgenden Codeausschnitt verwenden, um eine Tabelle zu erstellen, in der Magnetic Store Writes aktiviert sind. Mit Magnetic Store Writes können Sie Daten sowohl in das Aufbewahrungsfenster Ihres Speichers als auch in das Aufbewahrungsfenster Ihres Magnetspeichers schreiben.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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());
}
}
```
------
# Tabelle beschreiben
Sie können die folgenden Codefragmente verwenden, um Informationen zu den Attributen Ihrer Tabelle zu erhalten.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Im folgenden Codeausschnitt wird AWS SDK für JavaScript Version 3 verwendet. Weitere Informationen zur Installation und Verwendung des Clients finden Sie unter [Timestream Write Client — AWS SDK for v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html). JavaScript
Siehe auch [Klasse DescribeTableCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/describedatabasecommand.html) und [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;
}
}
```
Das folgende Snippet verwendet das AWS SDK für den JavaScript V2-Stil. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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());
}
}
```
------
# Tabelle aktualisieren
Sie können die folgenden Codefragmente verwenden, um eine Tabelle zu aktualisieren.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Im folgenden Codeausschnitt wird AWS SDK für JavaScript Version 3 verwendet. Weitere Informationen zur Installation und Verwendung des Clients finden Sie unter [Timestream Write Client — AWS SDK for v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html). JavaScript
Siehe auch [Klasse UpdateTableCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/updatetablecommand.html) und [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);
}
```
Das folgende Snippet verwendet das AWS SDK für den JavaScript V2-Stil. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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());
}
}
```
------
# Tabelle löschen
Sie können die folgenden Codefragmente verwenden, um eine Tabelle zu löschen.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Im folgenden Codeausschnitt wird AWS SDK für JavaScript Version 3 verwendet. Weitere Informationen zur Installation und Verwendung des Clients finden Sie unter [Timestream Write Client — AWS SDK for v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html). JavaScript
Siehe auch [Klasse DeleteTableCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/deletetablecommand.html) und [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;
}
}
```
Das folgende Snippet verwendet das AWS SDK für den JavaScript V2-Stil. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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());
}
}
```
------
# Auflisten von Tabellen
Sie können die folgenden Codefragmente verwenden, um Tabellen aufzulisten.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Im folgenden Codeausschnitt wird AWS SDK für JavaScript Version 3 verwendet. Weitere Informationen zur Installation und Verwendung des Clients finden Sie unter [Timestream Write Client — AWS SDK for v3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/index.html). JavaScript
Siehe auch [Klasse ListTablesCommand](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-timestream-write/classes/listtablescommand.html) und [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);
}
}
```
Das folgende Snippet verwendet das AWS SDK für den JavaScript V2-Stil. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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}");
}
```
------
# Daten schreiben (Einfügungen und Upserts)
**Topics**
+ [
## Stapel von Datensätzen schreiben
](#code-samples.write.write-batches)
+ [
## Schreiben von Datensatzstapeln mit gemeinsamen Attributen
](#code-samples.write.write-batches-common-attrs)
+ [
## Datensätze werden aktualisiert
](#code-samples.write.upserts)
+ [
## Beispiel für ein Attribut mit mehreren Kennzahlen
](#code-samples.write.data.multivalue)
+ [
## Behandlung von Schreibfehlern
](#code-samples.write.rejectedRecordException)
## Stapel von Datensätzen schreiben
Sie können die folgenden Codefragmente verwenden, um Daten in eine Amazon Timestream-Tabelle zu schreiben. Das Schreiben von Daten in Batches hilft dabei, die Schreibkosten zu optimieren. Weitere Informationen finden Sie unter [Berechnung der Anzahl der Schreibvorgänge](metering-and-pricing.writes.md#metering-and-pricing.writes.write-size-multiple-events).
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Der folgende Codeausschnitt verwendet das AWS SDK für JavaScript V2. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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());
}
}
```
------
## Schreiben von Datensatzstapeln mit gemeinsamen Attributen
Wenn Ihre Zeitreihendaten Kennzahlen und/oder Dimensionen aufweisen, die vielen Datenpunkten gemeinsam sind, können Sie auch die folgende optimierte Version der WriteRecords-API verwenden, um Daten in Timestream for einzufügen. LiveAnalytics Durch die Verwendung gemeinsamer Attribute bei der Batchverarbeitung können die Schreibkosten weiter optimiert werden, wie unter beschrieben. [Berechnung der Anzahl der Schreibvorgänge](metering-and-pricing.writes.md#metering-and-pricing.writes.write-size-multiple-events)
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Der folgende Codeausschnitt verwendet das AWS SDK für JavaScript V2. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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());
}
}
```
------
## Datensätze werden aktualisiert
Während die Standardschreibvorgänge in Amazon Timestream der Semantik des *ersten Writers* folgen, bei der Daten nur als Anhängen gespeichert werden und doppelte Datensätze zurückgewiesen werden, gibt es Anwendungen, für die die Fähigkeit erforderlich ist, Daten mithilfe der Semantik des *letzten Writers* in Amazon Timestream zu schreiben, bei denen der Datensatz mit der höchsten Version im System gespeichert wird. Es gibt auch Anwendungen, die die Möglichkeit erfordern, bestehende Datensätze zu aktualisieren. Um diesen Szenarien gerecht zu werden, bietet Amazon Timestream die Möglichkeit, Daten zu *ändern.* Upsert ist eine Operation, die einen Datensatz in das System einfügt, wenn der Datensatz nicht existiert, oder den Datensatz aktualisiert, falls einer existiert.
Sie können Datensätze aktualisieren, indem Sie sie beim Senden einer `Version` `WriteRecords` Anfrage in die Datensatzdefinition aufnehmen. Amazon Timestream speichert den Datensatz mit dem höchsten Wert`Version`. Das folgende Codebeispiel zeigt, wie Sie Daten ändern können:
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Der folgende Codeausschnitt verwendet das AWS SDK für JavaScript V2. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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());
}
}
```
------
## Beispiel für ein Attribut mit mehreren Kennzahlen
Dieses Beispiel veranschaulicht das Schreiben von Attributen mit mehreren Kennzahlen. [Attribute mit mehreren Kennzahlen](data-modeling.md#data-modeling-multiVsinglerecords) sind nützlich, wenn ein Gerät oder eine Anwendung, die Sie verfolgen, mehrere Messwerte oder Ereignisse gleichzeitig ausgibt.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Der folgende Codeausschnitt verwendet das AWS SDK für JavaScript V2. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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());
}
}
}
}
```
------
## Behandlung von Schreibfehlern
Schreibvorgänge in Amazon Timestream können aus einem oder mehreren der folgenden Gründe fehlschlagen:
+ Es gibt Datensätze mit Zeitstempeln, die außerhalb der Aufbewahrungsdauer des Speicherspeichers liegen.
+ Es gibt Datensätze, die Dimensionen und/oder Kennzahlen enthalten, die die von Timestream definierten Grenzwerte überschreiten.
+ Amazon Timestream hat doppelte Datensätze erkannt. Datensätze werden als doppelt markiert, wenn es mehrere Datensätze mit denselben Dimensionen, Zeitstempeln und Kennzahlnamen gibt, aber:
+ Die Kennzahlwerte sind unterschiedlich.
+ Version ist in der Anforderung nicht vorhanden, oder der Wert der Version im neuen Datensatz ist gleich oder niedriger als der vorhandene Wert. Wenn Amazon Timestream Daten aus diesem Grund ablehnt, enthält das `ExistingVersion` Feld in der die `RejectedRecords` aktuelle Version des Datensatzes, wie sie in Amazon Timestream gespeichert ist. Um eine Aktualisierung zu erzwingen, können Sie die Anfrage erneut senden, wobei eine Version für den Datensatz auf einen Wert gesetzt ist, der größer als der ist. `ExistingVersion`
Weitere Informationen zu Fehlern und abgelehnten Datensätzen finden Sie unter [Fehler](https://docs.aws.amazon.com/timestream/latest/developerguide/API_WriteRecords.html#API_WriteRecords_Errors) und [RejectedRecord](https://docs.aws.amazon.com/timestream/latest/developerguide/API_RejectedRecord.html).
Wenn Ihre Anwendung `RejectedRecordsException` beim Versuch, Datensätze in Timestream zu schreiben, eine Meldung erhält, können Sie die abgelehnten Datensätze analysieren, um mehr über die Schreibfehler zu erfahren, wie unten dargestellt.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Der folgende Codeausschnitt verwendet das AWS SDK für JavaScript V2. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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());
}
```
------
# Abfrage ausführen
**Topics**
+ [
## Ergebnisse paginieren
](#code-samples.run-query.pagination)
+ [
## Analysieren von Ergebnismengen
](#code-samples.run-query.parsing)
+ [
## Zugriff auf den Abfragestatus
](#code-samples.run-query.query-status)
## Ergebnisse paginieren
Wenn Sie eine Abfrage ausführen, gibt Timestream die Ergebnismenge paginiert zurück, um die Reaktionsfähigkeit Ihrer Anwendungen zu optimieren. Der folgende Codeausschnitt zeigt, wie Sie den Ergebnissatz paginieren können. Sie müssen alle Seiten der Ergebnismenge in einer Schleife durchgehen, bis Sie auf einen Nullwert stoßen. Paginierungstoken laufen 3 Stunden ab, nachdem sie von Timestream für ausgestellt wurden. LiveAnalytics
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Der folgende Codeausschnitt verwendet das AWS SDK für JavaScript V2. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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());
}
}
```
------
## Analysieren von Ergebnismengen
Sie können die folgenden Codefragmente verwenden, um Daten aus der Ergebnismenge zu extrahieren. Auf die Abfrageergebnisse kann bis zu 24 Stunden nach Abschluss einer Abfrage zugegriffen werden.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Der folgende Codeausschnitt verwendet das AWS SDK für JavaScript V2. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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)}]";
}
```
------
## Zugriff auf den Abfragestatus
Sie können auf den Abfragestatus zugreifen`QueryResponse`, der Informationen über den Fortschritt einer Abfrage, die von einer Abfrage gescannten Byte und die von einer Abfrage gemessenen Byte enthält. Die `bytesScanned` Werte `bytesMetered` und sind kumulativ und werden beim Paging der Abfrageergebnisse kontinuierlich aktualisiert. Sie können diese Informationen verwenden, um zu verstehen, welche Byte von einer einzelnen Abfrage gescannt wurden, und anhand dieser Informationen auch bestimmte Entscheidungen treffen. Wenn Sie beispielsweise annehmen, dass der Abfragepreis 0,01 USD pro gescanntem GB beträgt, möchten Sie möglicherweise Abfragen stornieren, die 25 USD pro Abfrage oder `X` GB überschreiten. Der folgende Codeausschnitt zeigt, wie das gemacht werden kann.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Der folgende Codeausschnitt verwendet das AWS SDK für JavaScript V2. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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());
}
}
```
------
Weitere Informationen zum Stornieren einer Abfrage finden Sie unter[Anfrage abbrechen](code-samples.cancel-query.md).
# UNLOAD-Abfrage ausführen
In den folgenden Codebeispielen wird eine UNLOAD-Abfrage aufgerufen. Weitere Informationen zu `UNLOAD` finden Sie unter [Verwenden von UNLOAD zum Exportieren von Abfrageergebnissen aus Timestream nach S3 für LiveAnalytics](export-unload.md). Beispiele für `UNLOAD` Abfragen finden Sie unter[Beispiel für einen Anwendungsfall für UNLOAD von Timestream für LiveAnalytics](export-unload-example-use-case.md).
**Topics**
+ [
## Erstellen Sie eine UNLOAD-Abfrage und führen Sie sie aus
](#code-samples.run-query-unload-build-and-run)
+ [
## Analysieren Sie die UNLOAD-Antwort und rufen Sie die Zeilenanzahl, den Manifest-Link und den Metadaten-Link ab
](#code-samples.run-query-unload-parse-response)
+ [
## Lesen und analysieren Sie den Inhalt des Manifests
](#code-samples.run-query-unload-parse-manifest)
+ [
## Lesen und analysieren Sie den Inhalt von Metadaten
](#code-samples.run-query-unload-parse-metadata)
## Erstellen Sie eine UNLOAD-Abfrage und führen Sie sie aus
------
#### [ 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
}
}
```
------
## Analysieren Sie die UNLOAD-Antwort und rufen Sie die Zeilenanzahl, den Manifest-Link und den Metadaten-Link ab
------
#### [ 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
}
```
------
## Lesen und analysieren Sie den Inhalt des Manifests
------
#### [ 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};
}
```
------
## Lesen und analysieren Sie den Inhalt von Metadaten
------
#### [ 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};
}
```
------
# Anfrage abbrechen
Sie können die folgenden Codefragmente verwenden, um eine Abfrage abzubrechen.
**Anmerkung**
Diese Codefragmente basieren auf vollständigen Beispielanwendungen auf. [GitHub](https://github.com/awslabs/amazon-timestream-tools/blob/master/sample_apps) Weitere Informationen zu den ersten Schritten mit den Beispielanwendungen finden Sie unter. [Beispielanwendung](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 ]
Der folgende Codeausschnitt verwendet das AWS SDK für JavaScript V2. Es basiert auf der Beispielanwendung unter [Node.js Beispiel Amazon Timestream für die LiveAnalytics Anwendung auf 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);
}
}
```
------
# Batch-Load-Aufgabe erstellen
Sie können die folgenden Codefragmente verwenden, um Batch-Load-Aufgaben zu erstellen.
------
#### [ 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 =