Doc AWS SDK ExamplesWord リポジトリには、さらに多くの GitHub の例があります。 AWS SDK
翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
an AWS SDK を使用した Amazon Keyspaces の基本を学ぶ
次のコード例は、以下を実行する方法を示しています。
キースペースとテーブルを作成します。テーブルスキーマは映画データを保持し、 point-in-time リカバリが有効になっています。
SigV4 認証を使用した安全な TLS 接続を使用してキースペースに接続します。
テーブルに対してクエリを実行します。ムービーデータを追加、取得、更新します。
テーブルを更新する。視聴したムービーを追跡する列を追加します。
テーブルを以前の状態に戻し、リソースをクリーンアップします。
- .NET
-
- AWS SDK for .NET
-
注記
GitHub には他にもあります。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 global using System.Security.Cryptography.X509Certificates; global using Amazon.Keyspaces; global using Amazon.Keyspaces.Model; global using KeyspacesActions; global using KeyspacesScenario; global using Microsoft.Extensions.Configuration; global using Microsoft.Extensions.DependencyInjection; global using Microsoft.Extensions.Hosting; global using Microsoft.Extensions.Logging; global using Microsoft.Extensions.Logging.Console; global using Microsoft.Extensions.Logging.Debug; global using Newtonsoft.Json; namespace KeyspacesBasics; /// <summary> /// Amazon Keyspaces (for Apache Cassandra) scenario. Shows some of the basic /// actions performed with Amazon Keyspaces. /// </summary> public class KeyspacesBasics { private static ILogger logger = null!; static async Task Main(string[] args) { // Set up dependency injection for the Amazon service. using var host = Host.CreateDefaultBuilder(args) .ConfigureLogging(logging => logging.AddFilter("System", LogLevel.Debug) .AddFilter<DebugLoggerProvider>("Microsoft", LogLevel.Information) .AddFilter<ConsoleLoggerProvider>("Microsoft", LogLevel.Trace)) .ConfigureServices((_, services) => services.AddAWSService<IAmazonKeyspaces>() .AddTransient<KeyspacesWrapper>() .AddTransient<CassandraWrapper>() ) .Build(); logger = LoggerFactory.Create(builder => { builder.AddConsole(); }) .CreateLogger<KeyspacesBasics>(); var configuration = new ConfigurationBuilder() .SetBasePath(Directory.GetCurrentDirectory()) .AddJsonFile("settings.json") // Load test settings from .json file. .AddJsonFile("settings.local.json", true) // Optionally load local settings. .Build(); var keyspacesWrapper = host.Services.GetRequiredService<KeyspacesWrapper>(); var uiMethods = new UiMethods(); var keyspaceName = configuration["KeyspaceName"]; var tableName = configuration["TableName"]; bool success; // Used to track the results of some operations. uiMethods.DisplayOverview(); uiMethods.PressEnter(); // Create the keyspace. var keyspaceArn = await keyspacesWrapper.CreateKeyspace(keyspaceName); // Wait for the keyspace to be available. GetKeyspace results in a // resource not found error until it is ready for use. try { var getKeyspaceArn = ""; Console.Write($"Created {keyspaceName}. Waiting for it to become available. "); do { getKeyspaceArn = await keyspacesWrapper.GetKeyspace(keyspaceName); Console.Write(". "); } while (getKeyspaceArn != keyspaceArn); } catch (ResourceNotFoundException) { Console.WriteLine("Waiting for keyspace to be created."); } Console.WriteLine($"\nThe keyspace {keyspaceName} is ready for use."); uiMethods.PressEnter(); // Create the table. // First define the schema. var allColumns = new List<ColumnDefinition> { new ColumnDefinition { Name = "title", Type = "text" }, new ColumnDefinition { Name = "year", Type = "int" }, new ColumnDefinition { Name = "release_date", Type = "timestamp" }, new ColumnDefinition { Name = "plot", Type = "text" }, }; var partitionKeys = new List<PartitionKey> { new PartitionKey { Name = "year", }, new PartitionKey { Name = "title" }, }; var tableSchema = new SchemaDefinition { AllColumns = allColumns, PartitionKeys = partitionKeys, }; var tableArn = await keyspacesWrapper.CreateTable(keyspaceName, tableSchema, tableName); // Wait for the table to be active. try { var resp = new GetTableResponse(); Console.Write("Waiting for the new table to be active. "); do { try { resp = await keyspacesWrapper.GetTable(keyspaceName, tableName); Console.Write("."); } catch (ResourceNotFoundException) { Console.Write("."); } } while (resp.Status != TableStatus.ACTIVE); // Display the table's schema. Console.WriteLine($"\nTable {tableName} has been created in {keyspaceName}"); Console.WriteLine("Let's take a look at the schema."); uiMethods.DisplayTitle("All columns"); resp.SchemaDefinition.AllColumns.ForEach(column => { Console.WriteLine($"{column.Name,-40}\t{column.Type,-20}"); }); uiMethods.DisplayTitle("Cluster keys"); resp.SchemaDefinition.ClusteringKeys.ForEach(clusterKey => { Console.WriteLine($"{clusterKey.Name,-40}\t{clusterKey.OrderBy,-20}"); }); uiMethods.DisplayTitle("Partition keys"); resp.SchemaDefinition.PartitionKeys.ForEach(partitionKey => { Console.WriteLine($"{partitionKey.Name}"); }); uiMethods.PressEnter(); } catch (ResourceNotFoundException ex) { Console.WriteLine($"Error: {ex.Message}"); } // Access Apache Cassandra using the Cassandra drive for C#. var cassandraWrapper = host.Services.GetRequiredService<CassandraWrapper>(); var movieFilePath = configuration["MovieFile"]; Console.WriteLine("Let's add some movies to the table we created."); var inserted = await cassandraWrapper.InsertIntoMovieTable(keyspaceName, tableName, movieFilePath); uiMethods.PressEnter(); Console.WriteLine("Added the following movies to the table:"); var rows = await cassandraWrapper.GetMovies(keyspaceName, tableName); uiMethods.DisplayTitle("All Movies"); foreach (var row in rows) { var title = row.GetValue<string>("title"); var year = row.GetValue<int>("year"); var plot = row.GetValue<string>("plot"); var release_date = row.GetValue<DateTime>("release_date"); Console.WriteLine($"{release_date}\t{title}\t{year}\n{plot}"); Console.WriteLine(uiMethods.SepBar); } // Update the table schema uiMethods.DisplayTitle("Update table schema"); Console.WriteLine("Now we will update the table to add a boolean field called watched."); // First save the current time as a UTC Date so the original // table can be restored later. var timeChanged = DateTime.UtcNow; // Now update the schema. var resourceArn = await keyspacesWrapper.UpdateTable(keyspaceName, tableName); uiMethods.PressEnter(); Console.WriteLine("Now let's mark some of the movies as watched."); // Pick some files to mark as watched. var movieToWatch = rows[2].GetValue<string>("title"); var watchedMovieYear = rows[2].GetValue<int>("year"); var changedRows = await cassandraWrapper.MarkMovieAsWatched(keyspaceName, tableName, movieToWatch, watchedMovieYear); movieToWatch = rows[6].GetValue<string>("title"); watchedMovieYear = rows[6].GetValue<int>("year"); changedRows = await cassandraWrapper.MarkMovieAsWatched(keyspaceName, tableName, movieToWatch, watchedMovieYear); movieToWatch = rows[9].GetValue<string>("title"); watchedMovieYear = rows[9].GetValue<int>("year"); changedRows = await cassandraWrapper.MarkMovieAsWatched(keyspaceName, tableName, movieToWatch, watchedMovieYear); movieToWatch = rows[10].GetValue<string>("title"); watchedMovieYear = rows[10].GetValue<int>("year"); changedRows = await cassandraWrapper.MarkMovieAsWatched(keyspaceName, tableName, movieToWatch, watchedMovieYear); movieToWatch = rows[13].GetValue<string>("title"); watchedMovieYear = rows[13].GetValue<int>("year"); changedRows = await cassandraWrapper.MarkMovieAsWatched(keyspaceName, tableName, movieToWatch, watchedMovieYear); uiMethods.DisplayTitle("Watched movies"); Console.WriteLine("These movies have been marked as watched:"); rows = await cassandraWrapper.GetWatchedMovies(keyspaceName, tableName); foreach (var row in rows) { var title = row.GetValue<string>("title"); var year = row.GetValue<int>("year"); Console.WriteLine($"{title,-40}\t{year,8}"); } uiMethods.PressEnter(); Console.WriteLine("We can restore the table to its previous state but that can take up to 20 minutes to complete."); string answer; do { Console.WriteLine("Do you want to restore the table? (y/n)"); answer = Console.ReadLine(); } while (answer.ToLower() != "y" && answer.ToLower() != "n"); if (answer == "y") { var restoredTableName = $"{tableName}_restored"; var restoredTableArn = await keyspacesWrapper.RestoreTable( keyspaceName, tableName, restoredTableName, timeChanged); // Loop and call GetTable until the table is gone. Once it has been // deleted completely, GetTable will raise a ResourceNotFoundException. bool wasRestored = false; try { do { var resp = await keyspacesWrapper.GetTable(keyspaceName, restoredTableName); wasRestored = (resp.Status == TableStatus.ACTIVE); } while (!wasRestored); } catch (ResourceNotFoundException) { // If the restored table raised an error, it isn't // ready yet. Console.Write("."); } } uiMethods.DisplayTitle("Clean up resources."); // Delete the table. success = await keyspacesWrapper.DeleteTable(keyspaceName, tableName); Console.WriteLine($"Table {tableName} successfully deleted from {keyspaceName}."); Console.WriteLine("Waiting for the table to be removed completely. "); // Loop and call GetTable until the table is gone. Once it has been // deleted completely, GetTable will raise a ResourceNotFoundException. bool wasDeleted = false; try { do { var resp = await keyspacesWrapper.GetTable(keyspaceName, tableName); } while (!wasDeleted); } catch (ResourceNotFoundException ex) { wasDeleted = true; Console.WriteLine($"{ex.Message} indicates that the table has been deleted."); } // Delete the keyspace. success = await keyspacesWrapper.DeleteKeyspace(keyspaceName); Console.WriteLine("The keyspace has been deleted and the demo is now complete."); } }
namespace KeyspacesActions; /// <summary> /// Performs Amazon Keyspaces (for Apache Cassandra) actions. /// </summary> public class KeyspacesWrapper { private readonly IAmazonKeyspaces _amazonKeyspaces; /// <summary> /// Constructor for the KeyspaceWrapper. /// </summary> /// <param name="amazonKeyspaces">An Amazon Keyspaces client object.</param> public KeyspacesWrapper(IAmazonKeyspaces amazonKeyspaces) { _amazonKeyspaces = amazonKeyspaces; } /// <summary> /// Create a new keyspace. /// </summary> /// <param name="keyspaceName">The name for the new keyspace.</param> /// <returns>The Amazon Resource Name (ARN) of the new keyspace.</returns> public async Task<string> CreateKeyspace(string keyspaceName) { var response = await _amazonKeyspaces.CreateKeyspaceAsync( new CreateKeyspaceRequest { KeyspaceName = keyspaceName }); return response.ResourceArn; } /// <summary> /// Create a new Amazon Keyspaces table. /// </summary> /// <param name="keyspaceName">The keyspace where the table will be created.</param> /// <param name="schema">The schema for the new table.</param> /// <param name="tableName">The name of the new table.</param> /// <returns>The Amazon Resource Name (ARN) of the new table.</returns> public async Task<string> CreateTable(string keyspaceName, SchemaDefinition schema, string tableName) { var request = new CreateTableRequest { KeyspaceName = keyspaceName, SchemaDefinition = schema, TableName = tableName, PointInTimeRecovery = new PointInTimeRecovery { Status = PointInTimeRecoveryStatus.ENABLED } }; var response = await _amazonKeyspaces.CreateTableAsync(request); return response.ResourceArn; } /// <summary> /// Delete an existing keyspace. /// </summary> /// <param name="keyspaceName"></param> /// <returns>A Boolean value indicating the success of the action.</returns> public async Task<bool> DeleteKeyspace(string keyspaceName) { var response = await _amazonKeyspaces.DeleteKeyspaceAsync( new DeleteKeyspaceRequest { KeyspaceName = keyspaceName }); return response.HttpStatusCode == HttpStatusCode.OK; } /// <summary> /// Delete an Amazon Keyspaces table. /// </summary> /// <param name="keyspaceName">The keyspace containing the table.</param> /// <param name="tableName">The name of the table to delete.</param> /// <returns>A Boolean value indicating the success of the action.</returns> public async Task<bool> DeleteTable(string keyspaceName, string tableName) { var response = await _amazonKeyspaces.DeleteTableAsync( new DeleteTableRequest { KeyspaceName = keyspaceName, TableName = tableName }); return response.HttpStatusCode == HttpStatusCode.OK; } /// <summary> /// Get data about a keyspace. /// </summary> /// <param name="keyspaceName">The name of the keyspace.</param> /// <returns>The Amazon Resource Name (ARN) of the keyspace.</returns> public async Task<string> GetKeyspace(string keyspaceName) { var response = await _amazonKeyspaces.GetKeyspaceAsync( new GetKeyspaceRequest { KeyspaceName = keyspaceName }); return response.ResourceArn; } /// <summary> /// Get information about an Amazon Keyspaces table. /// </summary> /// <param name="keyspaceName">The keyspace containing the table.</param> /// <param name="tableName">The name of the Amazon Keyspaces table.</param> /// <returns>The response containing data about the table.</returns> public async Task<GetTableResponse> GetTable(string keyspaceName, string tableName) { var response = await _amazonKeyspaces.GetTableAsync( new GetTableRequest { KeyspaceName = keyspaceName, TableName = tableName }); return response; } /// <summary> /// Lists all keyspaces for the account. /// </summary> /// <returns>Async task.</returns> public async Task ListKeyspaces() { var paginator = _amazonKeyspaces.Paginators.ListKeyspaces(new ListKeyspacesRequest()); Console.WriteLine("{0, -30}\t{1}", "Keyspace name", "Keyspace ARN"); Console.WriteLine(new string('-', Console.WindowWidth)); await foreach (var keyspace in paginator.Keyspaces) { Console.WriteLine($"{keyspace.KeyspaceName,-30}\t{keyspace.ResourceArn}"); } } /// <summary> /// Lists the Amazon Keyspaces tables in a keyspace. /// </summary> /// <param name="keyspaceName">The name of the keyspace.</param> /// <returns>A list of TableSummary objects.</returns> public async Task<List<TableSummary>> ListTables(string keyspaceName) { var response = await _amazonKeyspaces.ListTablesAsync(new ListTablesRequest { KeyspaceName = keyspaceName }); response.Tables.ForEach(table => { Console.WriteLine($"{table.KeyspaceName}\t{table.TableName}\t{table.ResourceArn}"); }); return response.Tables; } /// <summary> /// Restores the specified table to the specified point in time. /// </summary> /// <param name="keyspaceName">The keyspace containing the table.</param> /// <param name="tableName">The name of the table to restore.</param> /// <param name="timestamp">The time to which the table will be restored.</param> /// <returns>The Amazon Resource Name (ARN) of the restored table.</returns> public async Task<string> RestoreTable(string keyspaceName, string tableName, string restoredTableName, DateTime timestamp) { var request = new RestoreTableRequest { RestoreTimestamp = timestamp, SourceKeyspaceName = keyspaceName, SourceTableName = tableName, TargetKeyspaceName = keyspaceName, TargetTableName = restoredTableName }; var response = await _amazonKeyspaces.RestoreTableAsync(request); return response.RestoredTableARN; } /// <summary> /// Updates the movie table to add a boolean column named watched. /// </summary> /// <param name="keyspaceName">The keyspace containing the table.</param> /// <param name="tableName">The name of the table to change.</param> /// <returns>The Amazon Resource Name (ARN) of the updated table.</returns> public async Task<string> UpdateTable(string keyspaceName, string tableName) { var newColumn = new ColumnDefinition { Name = "watched", Type = "boolean" }; var request = new UpdateTableRequest { KeyspaceName = keyspaceName, TableName = tableName, AddColumns = new List<ColumnDefinition> { newColumn } }; var response = await _amazonKeyspaces.UpdateTableAsync(request); return response.ResourceArn; } }
using System.Net; using Cassandra; namespace KeyspacesScenario; /// <summary> /// Class to perform CRUD methods on an Amazon Keyspaces (for Apache Cassandra) database. /// /// NOTE: This sample uses a plain text authenticator for example purposes only. /// Recommended best practice is to use a SigV4 authentication plugin, if available. /// </summary> public class CassandraWrapper { private readonly IConfiguration _configuration; private readonly string _localPathToFile; private const string _certLocation = "https://certs.secureserver.net/repository/sf-class2-root.crt"; private const string _certFileName = "sf-class2-root.crt"; private readonly X509Certificate2Collection _certCollection; private X509Certificate2 _amazoncert; private Cluster _cluster; // User name and password for the service. private string _userName = null!; private string _pwd = null!; public CassandraWrapper() { _configuration = new ConfigurationBuilder() .SetBasePath(Directory.GetCurrentDirectory()) .AddJsonFile("settings.json") // Load test settings from .json file. .AddJsonFile("settings.local.json", true) // Optionally load local settings. .Build(); _localPathToFile = Path.GetTempPath(); // Get the Starfield digital certificate and save it locally. var client = new WebClient(); client.DownloadFile(_certLocation, $"{_localPathToFile}/{_certFileName}"); //var httpClient = new HttpClient(); //var httpResult = httpClient.Get(fileUrl); //using var resultStream = await httpResult.Content.ReadAsStreamAsync(); //using var fileStream = File.Create(pathToSave); //resultStream.CopyTo(fileStream); _certCollection = new X509Certificate2Collection(); _amazoncert = new X509Certificate2($"{_localPathToFile}/{_certFileName}"); // Get the user name and password stored in the configuration file. _userName = _configuration["UserName"]!; _pwd = _configuration["Password"]!; // For a list of Service Endpoints for Amazon Keyspaces, see: // https://docs.aws.amazon.com/keyspaces/latest/devguide/programmatic.endpoints.html var awsEndpoint = _configuration["ServiceEndpoint"]; _cluster = Cluster.Builder() .AddContactPoints(awsEndpoint) .WithPort(9142) .WithAuthProvider(new PlainTextAuthProvider(_userName, _pwd)) .WithSSL(new SSLOptions().SetCertificateCollection(_certCollection)) .WithQueryOptions( new QueryOptions() .SetConsistencyLevel(ConsistencyLevel.LocalQuorum) .SetSerialConsistencyLevel(ConsistencyLevel.LocalSerial)) .Build(); } /// <summary> /// Loads the contents of a JSON file into a list of movies to be /// added to the Apache Cassandra table. /// </summary> /// <param name="movieFileName">The full path to the JSON file.</param> /// <returns>A list of movie objects.</returns> public List<Movie> ImportMoviesFromJson(string movieFileName, int numToImport = 0) { if (!File.Exists(movieFileName)) { return null!; } using var sr = new StreamReader(movieFileName); string json = sr.ReadToEnd(); var allMovies = JsonConvert.DeserializeObject<List<Movie>>(json); // If numToImport = 0, return all movies in the collection. if (numToImport == 0) { // Now return the entire list of movies. return allMovies; } else { // Now return the first numToImport entries. return allMovies.GetRange(0, numToImport); } } /// <summary> /// Insert movies into the movie table. /// </summary> /// <param name="keyspaceName">The keyspace containing the table.</param> /// <param name="movieTableName">The Amazon Keyspaces table.</param> /// <param name="movieFilePath">The path to the resource file containing /// movie data to insert into the table.</param> /// <returns>A Boolean value indicating the success of the action.</returns> public async Task<bool> InsertIntoMovieTable(string keyspaceName, string movieTableName, string movieFilePath, int numToImport = 20) { // Get some movie data from the movies.json file var movies = ImportMoviesFromJson(movieFilePath, numToImport); var session = _cluster.Connect(keyspaceName); string insertCql; RowSet rs; // Now we insert the numToImport movies into the table. foreach (var movie in movies) { // Escape single quote characters in the plot. insertCql = $"INSERT INTO {keyspaceName}.{movieTableName} (title, year, release_date, plot) values($${movie.Title}$$, {movie.Year}, '{movie.Info.Release_Date.ToString("yyyy-MM-dd")}', $${movie.Info.Plot}$$)"; rs = await session.ExecuteAsync(new SimpleStatement(insertCql)); } return true; } /// <summary> /// Gets all of the movies in the movies table. /// </summary> /// <param name="keyspaceName">The keyspace containing the table.</param> /// <param name="tableName">The name of the table.</param> /// <returns>A list of row objects containing movie data.</returns> public async Task<List<Row>> GetMovies(string keyspaceName, string tableName) { var session = _cluster.Connect(); RowSet rs; try { rs = await session.ExecuteAsync(new SimpleStatement($"SELECT * FROM {keyspaceName}.{tableName}")); // Extract the row data from the returned RowSet. var rows = rs.GetRows().ToList(); return rows; } catch (Exception ex) { Console.WriteLine(ex.Message); return null!; } } /// <summary> /// Mark a movie in the movie table as watched. /// </summary> /// <param name="keyspaceName">The keyspace containing the table.</param> /// <param name="tableName">The name of the table.</param> /// <param name="title">The title of the movie to mark as watched.</param> /// <param name="year">The year the movie was released.</param> /// <returns>A set of rows containing the changed data.</returns> public async Task<List<Row>> MarkMovieAsWatched(string keyspaceName, string tableName, string title, int year) { var session = _cluster.Connect(); string updateCql = $"UPDATE {keyspaceName}.{tableName} SET watched=true WHERE title = $${title}$$ AND year = {year};"; var rs = await session.ExecuteAsync(new SimpleStatement(updateCql)); var rows = rs.GetRows().ToList(); return rows; } /// <summary> /// Retrieve the movies in the movies table where watched is true. /// </summary> /// <param name="keyspaceName">The keyspace containing the table.</param> /// <param name="tableName">The name of the table.</param> /// <returns>A list of row objects containing information about movies /// where watched is true.</returns> public async Task<List<Row>> GetWatchedMovies(string keyspaceName, string tableName) { var session = _cluster.Connect(); RowSet rs; try { rs = await session.ExecuteAsync(new SimpleStatement($"SELECT title, year, plot FROM {keyspaceName}.{tableName} WHERE watched = true ALLOW FILTERING")); // Extract the row data from the returned RowSet. var rows = rs.GetRows().ToList(); return rows; } catch (Exception ex) { Console.WriteLine(ex.Message); return null!; } } }
-
API の詳細については、AWS SDK for .NET API リファレンスの以下のトピックを参照してください。
-
- Java
-
- Java 2.x のSDK
-
注記
GitHub には他にもあります。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 /** * Before running this Java (v2) code example, set up your development * environment, including your credentials. * * For more information, see the following documentation topic: * * https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/get-started.html * * Before running this Java code example, you must create a * Java keystore (JKS) file and place it in your project's resources folder. * * This file is a secure file format used to hold certificate information for * Java applications. This is required to make a connection to Amazon Keyspaces. * For more information, see the following documentation topic: * * https://docs.aws.amazon.com/keyspaces/latest/devguide/using_java_driver.html * * This Java example performs the following tasks: * * 1. Create a keyspace. * 2. Check for keyspace existence. * 3. List keyspaces using a paginator. * 4. Create a table with a simple movie data schema and enable point-in-time * recovery. * 5. Check for the table to be in an Active state. * 6. List all tables in the keyspace. * 7. Use a Cassandra driver to insert some records into the Movie table. * 8. Get all records from the Movie table. * 9. Get a specific Movie. * 10. Get a UTC timestamp for the current time. * 11. Update the table schema to add a ‘watched’ Boolean column. * 12. Update an item as watched. * 13. Query for items with watched = True. * 14. Restore the table back to the previous state using the timestamp. * 15. Check for completion of the restore action. * 16. Delete the table. * 17. Confirm that both tables are deleted. * 18. Delete the keyspace. */ public class ScenarioKeyspaces { public static final String DASHES = new String(new char[80]).replace("\0", "-"); /* * Usage: * fileName - The name of the JSON file that contains movie data. (Get this file * from the GitHub repo at resources/sample_file.) * keyspaceName - The name of the keyspace to create. */ public static void main(String[] args) throws InterruptedException, IOException { String fileName = "<Replace with the JSON file that contains movie data>"; String keyspaceName = "<Replace with the name of the keyspace to create>"; String titleUpdate = "The Family"; int yearUpdate = 2013; String tableName = "Movie"; String tableNameRestore = "MovieRestore"; Region region = Region.US_EAST_1; KeyspacesClient keyClient = KeyspacesClient.builder() .region(region) .build(); DriverConfigLoader loader = DriverConfigLoader.fromClasspath("application.conf"); CqlSession session = CqlSession.builder() .withConfigLoader(loader) .build(); System.out.println(DASHES); System.out.println("Welcome to the Amazon Keyspaces example scenario."); System.out.println(DASHES); System.out.println(DASHES); System.out.println("1. Create a keyspace."); createKeySpace(keyClient, keyspaceName); System.out.println(DASHES); System.out.println(DASHES); Thread.sleep(5000); System.out.println("2. Check for keyspace existence."); checkKeyspaceExistence(keyClient, keyspaceName); System.out.println(DASHES); System.out.println(DASHES); System.out.println("3. List keyspaces using a paginator."); listKeyspacesPaginator(keyClient); System.out.println(DASHES); System.out.println(DASHES); System.out.println("4. Create a table with a simple movie data schema and enable point-in-time recovery."); createTable(keyClient, keyspaceName, tableName); System.out.println(DASHES); System.out.println(DASHES); System.out.println("5. Check for the table to be in an Active state."); Thread.sleep(6000); checkTable(keyClient, keyspaceName, tableName); System.out.println(DASHES); System.out.println(DASHES); System.out.println("6. List all tables in the keyspace."); listTables(keyClient, keyspaceName); System.out.println(DASHES); System.out.println(DASHES); System.out.println("7. Use a Cassandra driver to insert some records into the Movie table."); Thread.sleep(6000); loadData(session, fileName, keyspaceName); System.out.println(DASHES); System.out.println(DASHES); System.out.println("8. Get all records from the Movie table."); getMovieData(session, keyspaceName); System.out.println(DASHES); System.out.println(DASHES); System.out.println("9. Get a specific Movie."); getSpecificMovie(session, keyspaceName); System.out.println(DASHES); System.out.println(DASHES); System.out.println("10. Get a UTC timestamp for the current time."); ZonedDateTime utc = ZonedDateTime.now(ZoneOffset.UTC); System.out.println("DATETIME = " + Date.from(utc.toInstant())); System.out.println(DASHES); System.out.println(DASHES); System.out.println("11. Update the table schema to add a watched Boolean column."); updateTable(keyClient, keyspaceName, tableName); System.out.println(DASHES); System.out.println(DASHES); System.out.println("12. Update an item as watched."); Thread.sleep(10000); // Wait 10 secs for the update. updateRecord(session, keyspaceName, titleUpdate, yearUpdate); System.out.println(DASHES); System.out.println(DASHES); System.out.println("13. Query for items with watched = True."); getWatchedData(session, keyspaceName); System.out.println(DASHES); System.out.println(DASHES); System.out.println("14. Restore the table back to the previous state using the timestamp."); System.out.println("Note that the restore operation can take up to 20 minutes."); restoreTable(keyClient, keyspaceName, utc); System.out.println(DASHES); System.out.println(DASHES); System.out.println("15. Check for completion of the restore action."); Thread.sleep(5000); checkRestoredTable(keyClient, keyspaceName, "MovieRestore"); System.out.println(DASHES); System.out.println(DASHES); System.out.println("16. Delete both tables."); deleteTable(keyClient, keyspaceName, tableName); deleteTable(keyClient, keyspaceName, tableNameRestore); System.out.println(DASHES); System.out.println(DASHES); System.out.println("17. Confirm that both tables are deleted."); checkTableDelete(keyClient, keyspaceName, tableName); checkTableDelete(keyClient, keyspaceName, tableNameRestore); System.out.println(DASHES); System.out.println(DASHES); System.out.println("18. Delete the keyspace."); deleteKeyspace(keyClient, keyspaceName); System.out.println(DASHES); System.out.println(DASHES); System.out.println("The scenario has completed successfully."); System.out.println(DASHES); } public static void deleteKeyspace(KeyspacesClient keyClient, String keyspaceName) { try { DeleteKeyspaceRequest deleteKeyspaceRequest = DeleteKeyspaceRequest.builder() .keyspaceName(keyspaceName) .build(); keyClient.deleteKeyspace(deleteKeyspaceRequest); } catch (KeyspacesException e) { System.err.println(e.awsErrorDetails().errorMessage()); System.exit(1); } } public static void checkTableDelete(KeyspacesClient keyClient, String keyspaceName, String tableName) throws InterruptedException { try { String status; GetTableResponse response; GetTableRequest tableRequest = GetTableRequest.builder() .keyspaceName(keyspaceName) .tableName(tableName) .build(); // Keep looping until table cannot be found and a ResourceNotFoundException is // thrown. while (true) { response = keyClient.getTable(tableRequest); status = response.statusAsString(); System.out.println(". The table status is " + status); Thread.sleep(500); } } catch (ResourceNotFoundException e) { System.err.println(e.awsErrorDetails().errorMessage()); } System.out.println("The table is deleted"); } public static void deleteTable(KeyspacesClient keyClient, String keyspaceName, String tableName) { try { DeleteTableRequest tableRequest = DeleteTableRequest.builder() .keyspaceName(keyspaceName) .tableName(tableName) .build(); keyClient.deleteTable(tableRequest); } catch (KeyspacesException e) { System.err.println(e.awsErrorDetails().errorMessage()); System.exit(1); } } public static void checkRestoredTable(KeyspacesClient keyClient, String keyspaceName, String tableName) throws InterruptedException { try { boolean tableStatus = false; String status; GetTableResponse response = null; GetTableRequest tableRequest = GetTableRequest.builder() .keyspaceName(keyspaceName) .tableName(tableName) .build(); while (!tableStatus) { response = keyClient.getTable(tableRequest); status = response.statusAsString(); System.out.println("The table status is " + status); if (status.compareTo("ACTIVE") == 0) { tableStatus = true; } Thread.sleep(500); } List<ColumnDefinition> cols = response.schemaDefinition().allColumns(); for (ColumnDefinition def : cols) { System.out.println("The column name is " + def.name()); System.out.println("The column type is " + def.type()); } } catch (KeyspacesException e) { System.err.println(e.awsErrorDetails().errorMessage()); System.exit(1); } } public static void restoreTable(KeyspacesClient keyClient, String keyspaceName, ZonedDateTime utc) { try { Instant myTime = utc.toInstant(); RestoreTableRequest restoreTableRequest = RestoreTableRequest.builder() .restoreTimestamp(myTime) .sourceTableName("Movie") .targetKeyspaceName(keyspaceName) .targetTableName("MovieRestore") .sourceKeyspaceName(keyspaceName) .build(); RestoreTableResponse response = keyClient.restoreTable(restoreTableRequest); System.out.println("The ARN of the restored table is " + response.restoredTableARN()); } catch (KeyspacesException e) { System.err.println(e.awsErrorDetails().errorMessage()); System.exit(1); } } public static void getWatchedData(CqlSession session, String keyspaceName) { ResultSet resultSet = session .execute("SELECT * FROM \"" + keyspaceName + "\".\"Movie\" WHERE watched = true ALLOW FILTERING;"); resultSet.forEach(item -> { System.out.println("The Movie title is " + item.getString("title")); System.out.println("The Movie year is " + item.getInt("year")); System.out.println("The plot is " + item.getString("plot")); }); } public static void updateRecord(CqlSession session, String keySpace, String titleUpdate, int yearUpdate) { String sqlStatement = "UPDATE \"" + keySpace + "\".\"Movie\" SET watched=true WHERE title = :k0 AND year = :k1;"; BatchStatementBuilder builder = BatchStatement.builder(DefaultBatchType.UNLOGGED); builder.setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM); PreparedStatement preparedStatement = session.prepare(sqlStatement); builder.addStatement(preparedStatement.boundStatementBuilder() .setString("k0", titleUpdate) .setInt("k1", yearUpdate) .build()); BatchStatement batchStatement = builder.build(); session.execute(batchStatement); } public static void updateTable(KeyspacesClient keyClient, String keySpace, String tableName) { try { ColumnDefinition def = ColumnDefinition.builder() .name("watched") .type("boolean") .build(); UpdateTableRequest tableRequest = UpdateTableRequest.builder() .keyspaceName(keySpace) .tableName(tableName) .addColumns(def) .build(); keyClient.updateTable(tableRequest); } catch (KeyspacesException e) { System.err.println(e.awsErrorDetails().errorMessage()); System.exit(1); } } public static void getSpecificMovie(CqlSession session, String keyspaceName) { ResultSet resultSet = session.execute( "SELECT * FROM \"" + keyspaceName + "\".\"Movie\" WHERE title = 'The Family' ALLOW FILTERING ;"); resultSet.forEach(item -> { System.out.println("The Movie title is " + item.getString("title")); System.out.println("The Movie year is " + item.getInt("year")); System.out.println("The plot is " + item.getString("plot")); }); } // Get records from the Movie table. public static void getMovieData(CqlSession session, String keyspaceName) { ResultSet resultSet = session.execute("SELECT * FROM \"" + keyspaceName + "\".\"Movie\";"); resultSet.forEach(item -> { System.out.println("The Movie title is " + item.getString("title")); System.out.println("The Movie year is " + item.getInt("year")); System.out.println("The plot is " + item.getString("plot")); }); } // Load data into the table. public static void loadData(CqlSession session, String fileName, String keySpace) throws IOException { String sqlStatement = "INSERT INTO \"" + keySpace + "\".\"Movie\" (title, year, plot) values (:k0, :k1, :k2)"; JsonParser parser = new JsonFactory().createParser(new File(fileName)); com.fasterxml.jackson.databind.JsonNode rootNode = new ObjectMapper().readTree(parser); Iterator<JsonNode> iter = rootNode.iterator(); ObjectNode currentNode; int t = 0; while (iter.hasNext()) { // Add 20 movies to the table. if (t == 20) break; currentNode = (ObjectNode) iter.next(); int year = currentNode.path("year").asInt(); String title = currentNode.path("title").asText(); String plot = currentNode.path("info").path("plot").toString(); // Insert the data into the Amazon Keyspaces table. BatchStatementBuilder builder = BatchStatement.builder(DefaultBatchType.UNLOGGED); builder.setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM); PreparedStatement preparedStatement = session.prepare(sqlStatement); builder.addStatement(preparedStatement.boundStatementBuilder() .setString("k0", title) .setInt("k1", year) .setString("k2", plot) .build()); BatchStatement batchStatement = builder.build(); session.execute(batchStatement); t++; } System.out.println("You have added " + t + " records successfully!"); } public static void listTables(KeyspacesClient keyClient, String keyspaceName) { try { ListTablesRequest tablesRequest = ListTablesRequest.builder() .keyspaceName(keyspaceName) .build(); ListTablesIterable listRes = keyClient.listTablesPaginator(tablesRequest); listRes.stream() .flatMap(r -> r.tables().stream()) .forEach(content -> System.out.println(" ARN: " + content.resourceArn() + " Table name: " + content.tableName())); } catch (KeyspacesException e) { System.err.println(e.awsErrorDetails().errorMessage()); System.exit(1); } } public static void checkTable(KeyspacesClient keyClient, String keyspaceName, String tableName) throws InterruptedException { try { boolean tableStatus = false; String status; GetTableResponse response = null; GetTableRequest tableRequest = GetTableRequest.builder() .keyspaceName(keyspaceName) .tableName(tableName) .build(); while (!tableStatus) { response = keyClient.getTable(tableRequest); status = response.statusAsString(); System.out.println(". The table status is " + status); if (status.compareTo("ACTIVE") == 0) { tableStatus = true; } Thread.sleep(500); } List<ColumnDefinition> cols = response.schemaDefinition().allColumns(); for (ColumnDefinition def : cols) { System.out.println("The column name is " + def.name()); System.out.println("The column type is " + def.type()); } } catch (KeyspacesException e) { System.err.println(e.awsErrorDetails().errorMessage()); System.exit(1); } } public static void createTable(KeyspacesClient keyClient, String keySpace, String tableName) { try { // Set the columns. ColumnDefinition defTitle = ColumnDefinition.builder() .name("title") .type("text") .build(); ColumnDefinition defYear = ColumnDefinition.builder() .name("year") .type("int") .build(); ColumnDefinition defReleaseDate = ColumnDefinition.builder() .name("release_date") .type("timestamp") .build(); ColumnDefinition defPlot = ColumnDefinition.builder() .name("plot") .type("text") .build(); List<ColumnDefinition> colList = new ArrayList<>(); colList.add(defTitle); colList.add(defYear); colList.add(defReleaseDate); colList.add(defPlot); // Set the keys. PartitionKey yearKey = PartitionKey.builder() .name("year") .build(); PartitionKey titleKey = PartitionKey.builder() .name("title") .build(); List<PartitionKey> keyList = new ArrayList<>(); keyList.add(yearKey); keyList.add(titleKey); SchemaDefinition schemaDefinition = SchemaDefinition.builder() .partitionKeys(keyList) .allColumns(colList) .build(); PointInTimeRecovery timeRecovery = PointInTimeRecovery.builder() .status(PointInTimeRecoveryStatus.ENABLED) .build(); CreateTableRequest tableRequest = CreateTableRequest.builder() .keyspaceName(keySpace) .tableName(tableName) .schemaDefinition(schemaDefinition) .pointInTimeRecovery(timeRecovery) .build(); CreateTableResponse response = keyClient.createTable(tableRequest); System.out.println("The table ARN is " + response.resourceArn()); } catch (KeyspacesException e) { System.err.println(e.awsErrorDetails().errorMessage()); System.exit(1); } } public static void listKeyspacesPaginator(KeyspacesClient keyClient) { try { ListKeyspacesRequest keyspacesRequest = ListKeyspacesRequest.builder() .maxResults(10) .build(); ListKeyspacesIterable listRes = keyClient.listKeyspacesPaginator(keyspacesRequest); listRes.stream() .flatMap(r -> r.keyspaces().stream()) .forEach(content -> System.out.println(" Name: " + content.keyspaceName())); } catch (KeyspacesException e) { System.err.println(e.awsErrorDetails().errorMessage()); System.exit(1); } } public static void checkKeyspaceExistence(KeyspacesClient keyClient, String keyspaceName) { try { GetKeyspaceRequest keyspaceRequest = GetKeyspaceRequest.builder() .keyspaceName(keyspaceName) .build(); GetKeyspaceResponse response = keyClient.getKeyspace(keyspaceRequest); String name = response.keyspaceName(); System.out.println("The " + name + " KeySpace is ready"); } catch (KeyspacesException e) { System.err.println(e.awsErrorDetails().errorMessage()); System.exit(1); } } public static void createKeySpace(KeyspacesClient keyClient, String keyspaceName) { try { CreateKeyspaceRequest keyspaceRequest = CreateKeyspaceRequest.builder() .keyspaceName(keyspaceName) .build(); CreateKeyspaceResponse response = keyClient.createKeyspace(keyspaceRequest); System.out.println("The ARN of the KeySpace is " + response.resourceArn()); } catch (KeyspacesException e) { System.err.println(e.awsErrorDetails().errorMessage()); System.exit(1); } } }
-
API の詳細については、AWS SDK for Java 2.x API リファレンスの以下のトピックを参照してください。
-
- Kotlin
-
- Kotlin のSDK
-
注記
GitHub には他にもあります。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 /** Before running this Kotlin code example, set up your development environment, including your credentials. For more information, see the following documentation topic: https://docs.aws.amazon.com/sdk-for-kotlin/latest/developer-guide/setup.html This example uses a secure file format to hold certificate information for Kotlin applications. This is required to make a connection to Amazon Keyspaces. For more information, see the following documentation topic: https://docs.aws.amazon.com/keyspaces/latest/devguide/using_java_driver.html This Kotlin example performs the following tasks: 1. Create a keyspace. 2. Check for keyspace existence. 3. List keyspaces using a paginator. 4. Create a table with a simple movie data schema and enable point-in-time recovery. 5. Check for the table to be in an Active state. 6. List all tables in the keyspace. 7. Use a Cassandra driver to insert some records into the Movie table. 8. Get all records from the Movie table. 9. Get a specific Movie. 10. Get a UTC timestamp for the current time. 11. Update the table schema to add a ‘watched’ Boolean column. 12. Update an item as watched. 13. Query for items with watched = True. 14. Restore the table back to the previous state using the timestamp. 15. Check for completion of the restore action. 16. Delete the table. 17. Confirm that both tables are deleted. 18. Delete the keyspace. */ /* Usage: fileName - The name of the JSON file that contains movie data. (Get this file from the GitHub repo at resources/sample_file.) keyspaceName - The name of the keyspace to create. */ val DASHES: String = String(CharArray(80)).replace("\u0000", "-") suspend fun main() { val fileName = "<Replace with the JSON file that contains movie data>" val keyspaceName = "<Replace with the name of the keyspace to create>" val titleUpdate = "The Family" val yearUpdate = 2013 val tableName = "MovieKotlin" val tableNameRestore = "MovieRestore" val loader = DriverConfigLoader.fromClasspath("application.conf") val session = CqlSession .builder() .withConfigLoader(loader) .build() println(DASHES) println("Welcome to the Amazon Keyspaces example scenario.") println(DASHES) println(DASHES) println("1. Create a keyspace.") createKeySpace(keyspaceName) println(DASHES) println(DASHES) delay(5000) println("2. Check for keyspace existence.") checkKeyspaceExistence(keyspaceName) println(DASHES) println(DASHES) println("3. List keyspaces using a paginator.") listKeyspacesPaginator() println(DASHES) println(DASHES) println("4. Create a table with a simple movie data schema and enable point-in-time recovery.") createTable(keyspaceName, tableName) println(DASHES) println(DASHES) println("5. Check for the table to be in an Active state.") delay(6000) checkTable(keyspaceName, tableName) println(DASHES) println(DASHES) println("6. List all tables in the keyspace.") listTables(keyspaceName) println(DASHES) println(DASHES) println("7. Use a Cassandra driver to insert some records into the Movie table.") delay(6000) loadData(session, fileName, keyspaceName) println(DASHES) println(DASHES) println("8. Get all records from the Movie table.") getMovieData(session, keyspaceName) println(DASHES) println(DASHES) println("9. Get a specific Movie.") getSpecificMovie(session, keyspaceName) println(DASHES) println(DASHES) println("10. Get a UTC timestamp for the current time.") val utc = ZonedDateTime.now(ZoneOffset.UTC) println("DATETIME = ${Date.from(utc.toInstant())}") println(DASHES) println(DASHES) println("11. Update the table schema to add a watched Boolean column.") updateTable(keyspaceName, tableName) println(DASHES) println(DASHES) println("12. Update an item as watched.") delay(10000) // Wait 10 seconds for the update. updateRecord(session, keyspaceName, titleUpdate, yearUpdate) println(DASHES) println(DASHES) println("13. Query for items with watched = True.") getWatchedData(session, keyspaceName) println(DASHES) println(DASHES) println("14. Restore the table back to the previous state using the timestamp.") println("Note that the restore operation can take up to 20 minutes.") restoreTable(keyspaceName, utc) println(DASHES) println(DASHES) println("15. Check for completion of the restore action.") delay(5000) checkRestoredTable(keyspaceName, "MovieRestore") println(DASHES) println(DASHES) println("16. Delete both tables.") deleteTable(keyspaceName, tableName) deleteTable(keyspaceName, tableNameRestore) println(DASHES) println(DASHES) println("17. Confirm that both tables are deleted.") checkTableDelete(keyspaceName, tableName) checkTableDelete(keyspaceName, tableNameRestore) println(DASHES) println(DASHES) println("18. Delete the keyspace.") deleteKeyspace(keyspaceName) println(DASHES) println(DASHES) println("The scenario has completed successfully.") println(DASHES) } suspend fun deleteKeyspace(keyspaceNameVal: String?) { val deleteKeyspaceRequest = DeleteKeyspaceRequest { keyspaceName = keyspaceNameVal } KeyspacesClient { region = "us-east-1" }.use { keyClient -> keyClient.deleteKeyspace(deleteKeyspaceRequest) } } suspend fun checkTableDelete( keyspaceNameVal: String?, tableNameVal: String?, ) { var status: String var response: GetTableResponse val tableRequest = GetTableRequest { keyspaceName = keyspaceNameVal tableName = tableNameVal } try { KeyspacesClient { region = "us-east-1" }.use { keyClient -> // Keep looping until the table cannot be found and a ResourceNotFoundException is thrown. while (true) { response = keyClient.getTable(tableRequest) status = response.status.toString() println(". The table status is $status") delay(500) } } } catch (e: ResourceNotFoundException) { println(e.message) } println("The table is deleted") } suspend fun deleteTable( keyspaceNameVal: String?, tableNameVal: String?, ) { val tableRequest = DeleteTableRequest { keyspaceName = keyspaceNameVal tableName = tableNameVal } KeyspacesClient { region = "us-east-1" }.use { keyClient -> keyClient.deleteTable(tableRequest) } } suspend fun checkRestoredTable( keyspaceNameVal: String?, tableNameVal: String?, ) { var tableStatus = false var status: String var response: GetTableResponse? = null val tableRequest = GetTableRequest { keyspaceName = keyspaceNameVal tableName = tableNameVal } KeyspacesClient { region = "us-east-1" }.use { keyClient -> while (!tableStatus) { response = keyClient.getTable(tableRequest) status = response!!.status.toString() println("The table status is $status") if (status.compareTo("ACTIVE") == 0) { tableStatus = true } delay(500) } val cols = response!!.schemaDefinition?.allColumns if (cols != null) { for (def in cols) { println("The column name is ${def.name}") println("The column type is ${def.type}") } } } } suspend fun restoreTable( keyspaceName: String?, utc: ZonedDateTime, ) { // Create an aws.smithy.kotlin.runtime.time.Instant value. val timeStamp = aws.smithy.kotlin.runtime.time .Instant(utc.toInstant()) val restoreTableRequest = RestoreTableRequest { restoreTimestamp = timeStamp sourceTableName = "MovieKotlin" targetKeyspaceName = keyspaceName targetTableName = "MovieRestore" sourceKeyspaceName = keyspaceName } KeyspacesClient { region = "us-east-1" }.use { keyClient -> val response = keyClient.restoreTable(restoreTableRequest) println("The ARN of the restored table is ${response.restoredTableArn}") } } fun getWatchedData( session: CqlSession, keyspaceName: String, ) { val resultSet = session.execute("SELECT * FROM \"$keyspaceName\".\"MovieKotlin\" WHERE watched = true ALLOW FILTERING;") resultSet.forEach { item: Row -> println("The Movie title is ${item.getString("title")}") println("The Movie year is ${item.getInt("year")}") println("The plot is ${item.getString("plot")}") } } fun updateRecord( session: CqlSession, keySpace: String, titleUpdate: String?, yearUpdate: Int, ) { val sqlStatement = "UPDATE \"$keySpace\".\"MovieKotlin\" SET watched=true WHERE title = :k0 AND year = :k1;" val builder = BatchStatement.builder(DefaultBatchType.UNLOGGED) builder.setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM) val preparedStatement = session.prepare(sqlStatement) builder.addStatement( preparedStatement .boundStatementBuilder() .setString("k0", titleUpdate) .setInt("k1", yearUpdate) .build(), ) val batchStatement = builder.build() session.execute(batchStatement) } suspend fun updateTable( keySpace: String?, tableNameVal: String?, ) { val def = ColumnDefinition { name = "watched" type = "boolean" } val tableRequest = UpdateTableRequest { keyspaceName = keySpace tableName = tableNameVal addColumns = listOf(def) } KeyspacesClient { region = "us-east-1" }.use { keyClient -> keyClient.updateTable(tableRequest) } } fun getSpecificMovie( session: CqlSession, keyspaceName: String, ) { val resultSet = session.execute("SELECT * FROM \"$keyspaceName\".\"MovieKotlin\" WHERE title = 'The Family' ALLOW FILTERING ;") resultSet.forEach { item: Row -> println("The Movie title is ${item.getString("title")}") println("The Movie year is ${item.getInt("year")}") println("The plot is ${item.getString("plot")}") } } // Get records from the Movie table. fun getMovieData( session: CqlSession, keyspaceName: String, ) { val resultSet = session.execute("SELECT * FROM \"$keyspaceName\".\"MovieKotlin\";") resultSet.forEach { item: Row -> println("The Movie title is ${item.getString("title")}") println("The Movie year is ${item.getInt("year")}") println("The plot is ${item.getString("plot")}") } } // Load data into the table. fun loadData( session: CqlSession, fileName: String, keySpace: String, ) { val sqlStatement = "INSERT INTO \"$keySpace\".\"MovieKotlin\" (title, year, plot) values (:k0, :k1, :k2)" val parser = JsonFactory().createParser(File(fileName)) val rootNode = ObjectMapper().readTree<JsonNode>(parser) val iter: Iterator<JsonNode> = rootNode.iterator() var currentNode: ObjectNode var t = 0 while (iter.hasNext()) { if (t == 50) { break } currentNode = iter.next() as ObjectNode val year = currentNode.path("year").asInt() val title = currentNode.path("title").asText() val info = currentNode.path("info").toString() // Insert the data into the Amazon Keyspaces table. val builder = BatchStatement.builder(DefaultBatchType.UNLOGGED) builder.setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM) val preparedStatement: PreparedStatement = session.prepare(sqlStatement) builder.addStatement( preparedStatement .boundStatementBuilder() .setString("k0", title) .setInt("k1", year) .setString("k2", info) .build(), ) val batchStatement = builder.build() session.execute(batchStatement) t++ } } suspend fun listTables(keyspaceNameVal: String?) { val tablesRequest = ListTablesRequest { keyspaceName = keyspaceNameVal } KeyspacesClient { region = "us-east-1" }.use { keyClient -> keyClient .listTablesPaginated(tablesRequest) .transform { it.tables?.forEach { obj -> emit(obj) } } .collect { obj -> println(" ARN: ${obj.resourceArn} Table name: ${obj.tableName}") } } } suspend fun checkTable( keyspaceNameVal: String?, tableNameVal: String?, ) { var tableStatus = false var status: String var response: GetTableResponse? = null val tableRequest = GetTableRequest { keyspaceName = keyspaceNameVal tableName = tableNameVal } KeyspacesClient { region = "us-east-1" }.use { keyClient -> while (!tableStatus) { response = keyClient.getTable(tableRequest) status = response!!.status.toString() println(". The table status is $status") if (status.compareTo("ACTIVE") == 0) { tableStatus = true } delay(500) } val cols: List<ColumnDefinition>? = response!!.schemaDefinition?.allColumns if (cols != null) { for (def in cols) { println("The column name is ${def.name}") println("The column type is ${def.type}") } } } } suspend fun createTable( keySpaceVal: String?, tableNameVal: String?, ) { // Set the columns. val defTitle = ColumnDefinition { name = "title" type = "text" } val defYear = ColumnDefinition { name = "year" type = "int" } val defReleaseDate = ColumnDefinition { name = "release_date" type = "timestamp" } val defPlot = ColumnDefinition { name = "plot" type = "text" } val colList = ArrayList<ColumnDefinition>() colList.add(defTitle) colList.add(defYear) colList.add(defReleaseDate) colList.add(defPlot) // Set the keys. val yearKey = PartitionKey { name = "year" } val titleKey = PartitionKey { name = "title" } val keyList = ArrayList<PartitionKey>() keyList.add(yearKey) keyList.add(titleKey) val schemaDefinitionOb = SchemaDefinition { partitionKeys = keyList allColumns = colList } val timeRecovery = PointInTimeRecovery { status = PointInTimeRecoveryStatus.Enabled } val tableRequest = CreateTableRequest { keyspaceName = keySpaceVal tableName = tableNameVal schemaDefinition = schemaDefinitionOb pointInTimeRecovery = timeRecovery } KeyspacesClient { region = "us-east-1" }.use { keyClient -> val response = keyClient.createTable(tableRequest) println("The table ARN is ${response.resourceArn}") } } suspend fun listKeyspacesPaginator() { KeyspacesClient { region = "us-east-1" }.use { keyClient -> keyClient .listKeyspacesPaginated(ListKeyspacesRequest {}) .transform { it.keyspaces?.forEach { obj -> emit(obj) } } .collect { obj -> println("Name: ${obj.keyspaceName}") } } } suspend fun checkKeyspaceExistence(keyspaceNameVal: String?) { val keyspaceRequest = GetKeyspaceRequest { keyspaceName = keyspaceNameVal } KeyspacesClient { region = "us-east-1" }.use { keyClient -> val response: GetKeyspaceResponse = keyClient.getKeyspace(keyspaceRequest) val name = response.keyspaceName println("The $name KeySpace is ready") } } suspend fun createKeySpace(keyspaceNameVal: String) { val keyspaceRequest = CreateKeyspaceRequest { keyspaceName = keyspaceNameVal } KeyspacesClient { region = "us-east-1" }.use { keyClient -> val response = keyClient.createKeyspace(keyspaceRequest) println("The ARN of the KeySpace is ${response.resourceArn}") } }
-
API の詳細については、Word AWS for Kotlin SDK APIリファレンスの以下のトピックを参照してください。
-
- Python
-
- Python 用 SDK (Boto3)
-
注記
GitHub には他にもあります。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 コマンドプロンプトからインタラクティブのシナリオを実行します。
class KeyspaceScenario: """Runs an interactive scenario that shows how to get started using Amazon Keyspaces.""" def __init__(self, ks_wrapper): """ :param ks_wrapper: An object that wraps Amazon Keyspace actions. """ self.ks_wrapper = ks_wrapper @demo_func def create_keyspace(self): """ 1. Creates a keyspace. 2. Lists up to 10 keyspaces in your account. """ print("Let's create a keyspace.") ks_name = q.ask( "Enter a name for your new keyspace.\nThe name can contain only letters, " "numbers and underscores: ", q.non_empty, ) if self.ks_wrapper.exists_keyspace(ks_name): print(f"A keyspace named {ks_name} exists.") else: ks_arn = self.ks_wrapper.create_keyspace(ks_name) ks_exists = False while not ks_exists: wait(3) ks_exists = self.ks_wrapper.exists_keyspace(ks_name) print(f"Created a new keyspace.\n\t{ks_arn}.") print("The first 10 keyspaces in your account are:\n") self.ks_wrapper.list_keyspaces(10) @demo_func def create_table(self): """ 1. Creates a table in the keyspace. The table is configured with a schema to hold movie data and has point-in-time recovery enabled. 2. Waits for the table to be in an active state. 3. Displays schema information for the table. 4. Lists tables in the keyspace. """ print("Let's create a table for movies in your keyspace.") table_name = q.ask("Enter a name for your table: ", q.non_empty) table = self.ks_wrapper.get_table(table_name) if table is not None: print( f"A table named {table_name} already exists in keyspace " f"{self.ks_wrapper.ks_name}." ) else: table_arn = self.ks_wrapper.create_table(table_name) print(f"Created table {table_name}:\n\t{table_arn}") table = {"status": None} print("Waiting for your table to be ready...") while table["status"] != "ACTIVE": wait(5) table = self.ks_wrapper.get_table(table_name) print(f"Your table is {table['status']}. Its schema is:") pp(table["schemaDefinition"]) print("\nThe tables in your keyspace are:\n") self.ks_wrapper.list_tables() @demo_func def ensure_tls_cert(self): """ Ensures you have a TLS certificate available to use to secure the connection to the keyspace. This function downloads a default certificate or lets you specify your own. """ print("To connect to your keyspace, you must have a TLS certificate.") print("Checking for TLS certificate...") cert_path = os.path.join( os.path.dirname(__file__), QueryManager.DEFAULT_CERT_FILE ) if not os.path.exists(cert_path): cert_choice = q.ask( f"Press enter to download a certificate from {QueryManager.CERT_URL} " f"or enter the full path to the certificate you want to use: " ) if cert_choice: cert_path = cert_choice else: cert = requests.get(QueryManager.CERT_URL).text with open(cert_path, "w") as cert_file: cert_file.write(cert) else: q.ask(f"Certificate {cert_path} found. Press Enter to continue.") print( f"Certificate {cert_path} will be used to secure the connection to your keyspace." ) return cert_path @demo_func def query_table(self, qm, movie_file): """ 1. Adds movies to the table from a sample movie data file. 2. Gets a list of movies from the table and lets you select one. 3. Displays more information about the selected movie. """ qm.add_movies(self.ks_wrapper.table_name, movie_file) movies = qm.get_movies(self.ks_wrapper.table_name) print(f"Added {len(movies)} movies to the table:") sel = q.choose("Pick one to learn more about it: ", [m.title for m in movies]) movie_choice = qm.get_movie( self.ks_wrapper.table_name, movies[sel].title, movies[sel].year ) print(movie_choice.title) print(f"\tReleased: {movie_choice.release_date}") print(f"\tPlot: {movie_choice.plot}") @demo_func def update_and_restore_table(self, qm): """ 1. Updates the table by adding a column to track watched movies. 2. Marks some of the movies as watched. 3. Gets the list of watched movies from the table. 4. Restores to a movies_restored table at a previous point in time. 5. Gets the list of movies from the restored table. """ print("Let's add a column to record which movies you've watched.") pre_update_timestamp = datetime.utcnow() print( f"Recorded the current UTC time of {pre_update_timestamp} so we can restore the table later." ) self.ks_wrapper.update_table() print("Waiting for your table to update...") table = {"status": "UPDATING"} while table["status"] != "ACTIVE": wait(5) table = self.ks_wrapper.get_table(self.ks_wrapper.table_name) print("Column 'watched' added to table.") q.ask( "Let's mark some of the movies as watched. Press Enter when you're ready.\n" ) movies = qm.get_movies(self.ks_wrapper.table_name) for movie in movies[:10]: qm.watched_movie(self.ks_wrapper.table_name, movie.title, movie.year) print(f"Marked {movie.title} as watched.") movies = qm.get_movies(self.ks_wrapper.table_name, watched=True) print("-" * 88) print("The watched movies in our table are:\n") for movie in movies: print(movie.title) print("-" * 88) if q.ask( "Do you want to restore the table to the way it was before all of these\n" "updates? Keep in mind, this can take up to 20 minutes. (y/n) ", q.is_yesno, ): starting_table_name = self.ks_wrapper.table_name table_name_restored = self.ks_wrapper.restore_table(pre_update_timestamp) table = {"status": "RESTORING"} while table["status"] != "ACTIVE": wait(10) table = self.ks_wrapper.get_table(table_name_restored) print( f"Restored {starting_table_name} to {table_name_restored} " f"at a point in time of {pre_update_timestamp}." ) movies = qm.get_movies(table_name_restored) print("Now the movies in our table are:") for movie in movies: print(movie.title) def cleanup(self, cert_path): """ 1. Deletes the table and waits for it to be removed. 2. Deletes the keyspace. :param cert_path: The path of the TLS certificate used in the demo. If the certificate was downloaded during the demo, it is removed. """ if q.ask( f"Do you want to delete your {self.ks_wrapper.table_name} table and " f"{self.ks_wrapper.ks_name} keyspace? (y/n) ", q.is_yesno, ): table_name = self.ks_wrapper.table_name self.ks_wrapper.delete_table() table = self.ks_wrapper.get_table(table_name) print("Waiting for the table to be deleted.") while table is not None: wait(5) table = self.ks_wrapper.get_table(table_name) print("Table deleted.") self.ks_wrapper.delete_keyspace() print( "Keyspace deleted. If you chose to restore your table during the " "demo, the original table is also deleted." ) if cert_path == os.path.join( os.path.dirname(__file__), QueryManager.DEFAULT_CERT_FILE ) and os.path.exists(cert_path): os.remove(cert_path) print("Removed certificate that was downloaded for this demo.") def run_scenario(self): logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") print("-" * 88) print("Welcome to the Amazon Keyspaces (for Apache Cassandra) demo.") print("-" * 88) self.create_keyspace() self.create_table() cert_file_path = self.ensure_tls_cert() # Use a context manager to ensure the connection to the keyspace is closed. with QueryManager( cert_file_path, boto3.DEFAULT_SESSION, self.ks_wrapper.ks_name ) as qm: self.query_table(qm, "../../../resources/sample_files/movies.json") self.update_and_restore_table(qm) self.cleanup(cert_file_path) print("\nThanks for watching!") print("-" * 88) if __name__ == "__main__": try: scenario = KeyspaceScenario(KeyspaceWrapper.from_client()) scenario.run_scenario() except Exception: logging.exception("Something went wrong with the demo.")
キースペースとテーブルアクションをラップするクラスを定義します。
class KeyspaceWrapper: """Encapsulates Amazon Keyspaces (for Apache Cassandra) keyspace and table actions.""" def __init__(self, keyspaces_client): """ :param keyspaces_client: A Boto3 Amazon Keyspaces client. """ self.keyspaces_client = keyspaces_client self.ks_name = None self.ks_arn = None self.table_name = None @classmethod def from_client(cls): keyspaces_client = boto3.client("keyspaces") return cls(keyspaces_client) def create_keyspace(self, name): """ Creates a keyspace. :param name: The name to give the keyspace. :return: The Amazon Resource Name (ARN) of the new keyspace. """ try: response = self.keyspaces_client.create_keyspace(keyspaceName=name) self.ks_name = name self.ks_arn = response["resourceArn"] except ClientError as err: logger.error( "Couldn't create %s. Here's why: %s: %s", name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return self.ks_arn def exists_keyspace(self, name): """ Checks whether a keyspace exists. :param name: The name of the keyspace to look up. :return: True when the keyspace exists. Otherwise, False. """ try: response = self.keyspaces_client.get_keyspace(keyspaceName=name) self.ks_name = response["keyspaceName"] self.ks_arn = response["resourceArn"] exists = True except ClientError as err: if err.response["Error"]["Code"] == "ResourceNotFoundException": logger.info("Keyspace %s does not exist.", name) exists = False else: logger.error( "Couldn't verify %s exists. Here's why: %s: %s", name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise return exists def list_keyspaces(self, limit): """ Lists the keyspaces in your account. :param limit: The maximum number of keyspaces to list. """ try: ks_paginator = self.keyspaces_client.get_paginator("list_keyspaces") for page in ks_paginator.paginate(PaginationConfig={"MaxItems": limit}): for ks in page["keyspaces"]: print(ks["keyspaceName"]) print(f"\t{ks['resourceArn']}") except ClientError as err: logger.error( "Couldn't list keyspaces. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def create_table(self, table_name): """ Creates a table in the keyspace. The table is created with a schema for storing movie data and has point-in-time recovery enabled. :param table_name: The name to give the table. :return: The ARN of the new table. """ try: response = self.keyspaces_client.create_table( keyspaceName=self.ks_name, tableName=table_name, schemaDefinition={ "allColumns": [ {"name": "title", "type": "text"}, {"name": "year", "type": "int"}, {"name": "release_date", "type": "timestamp"}, {"name": "plot", "type": "text"}, ], "partitionKeys": [{"name": "year"}, {"name": "title"}], }, pointInTimeRecovery={"status": "ENABLED"}, ) except ClientError as err: logger.error( "Couldn't create table %s. Here's why: %s: %s", table_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response["resourceArn"] def get_table(self, table_name): """ Gets data about a table in the keyspace. :param table_name: The name of the table to look up. :return: Data about the table. """ try: response = self.keyspaces_client.get_table( keyspaceName=self.ks_name, tableName=table_name ) self.table_name = table_name except ClientError as err: if err.response["Error"]["Code"] == "ResourceNotFoundException": logger.info("Table %s does not exist.", table_name) self.table_name = None response = None else: logger.error( "Couldn't verify %s exists. Here's why: %s: %s", table_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise return response def list_tables(self): """ Lists the tables in the keyspace. """ try: table_paginator = self.keyspaces_client.get_paginator("list_tables") for page in table_paginator.paginate(keyspaceName=self.ks_name): for table in page["tables"]: print(table["tableName"]) print(f"\t{table['resourceArn']}") except ClientError as err: logger.error( "Couldn't list tables in keyspace %s. Here's why: %s: %s", self.ks_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def update_table(self): """ Updates the schema of the table. This example updates a table of movie data by adding a new column that tracks whether the movie has been watched. """ try: self.keyspaces_client.update_table( keyspaceName=self.ks_name, tableName=self.table_name, addColumns=[{"name": "watched", "type": "boolean"}], ) except ClientError as err: logger.error( "Couldn't update table %s. Here's why: %s: %s", self.table_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def restore_table(self, restore_timestamp): """ Restores the table to a previous point in time. The table is restored to a new table in the same keyspace. :param restore_timestamp: The point in time to restore the table. This time must be in UTC format. :return: The name of the restored table. """ try: restored_table_name = f"{self.table_name}_restored" self.keyspaces_client.restore_table( sourceKeyspaceName=self.ks_name, sourceTableName=self.table_name, targetKeyspaceName=self.ks_name, targetTableName=restored_table_name, restoreTimestamp=restore_timestamp, ) except ClientError as err: logger.error( "Couldn't restore table %s. Here's why: %s: %s", restore_timestamp, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return restored_table_name def delete_table(self): """ Deletes the table from the keyspace. """ try: self.keyspaces_client.delete_table( keyspaceName=self.ks_name, tableName=self.table_name ) self.table_name = None except ClientError as err: logger.error( "Couldn't delete table %s. Here's why: %s: %s", self.table_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def delete_keyspace(self): """ Deletes the keyspace. """ try: self.keyspaces_client.delete_keyspace(keyspaceName=self.ks_name) self.ks_name = None except ClientError as err: logger.error( "Couldn't delete keyspace %s. Here's why: %s: %s", self.ks_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise
キースペースへの TLS 接続を作成し、SigV4 で認証し、キースペースのテーブルに CQL クエリを送信するクラスを定義します。
class QueryManager: """ Manages queries to an Amazon Keyspaces (for Apache Cassandra) keyspace. Queries are secured by TLS and authenticated by using the Signature V4 (SigV4) AWS signing protocol. This is more secure than sending username and password with a plain-text authentication provider. This example downloads a default certificate to secure TLS, or lets you specify your own. This example uses a table of movie data to demonstrate basic queries. """ DEFAULT_CERT_FILE = "sf-class2-root.crt" CERT_URL = f"https://certs.secureserver.net/repository/sf-class2-root.crt" def __init__(self, cert_file_path, boto_session, keyspace_name): """ :param cert_file_path: The path and file name of the certificate used for TLS. :param boto_session: A Boto3 session. This is used to acquire your AWS credentials. :param keyspace_name: The name of the keyspace to connect. """ self.cert_file_path = cert_file_path self.boto_session = boto_session self.ks_name = keyspace_name self.cluster = None self.session = None def __enter__(self): """ Creates a session connection to the keyspace that is secured by TLS and authenticated by SigV4. """ ssl_context = SSLContext(PROTOCOL_TLSv1_2) ssl_context.load_verify_locations(self.cert_file_path) ssl_context.verify_mode = CERT_REQUIRED auth_provider = SigV4AuthProvider(self.boto_session) contact_point = f"cassandra.{self.boto_session.region_name}.amazonaws.com" exec_profile = ExecutionProfile( consistency_level=ConsistencyLevel.LOCAL_QUORUM, load_balancing_policy=DCAwareRoundRobinPolicy(), ) self.cluster = Cluster( [contact_point], ssl_context=ssl_context, auth_provider=auth_provider, port=9142, execution_profiles={EXEC_PROFILE_DEFAULT: exec_profile}, protocol_version=4, ) self.cluster.__enter__() self.session = self.cluster.connect(self.ks_name) return self def __exit__(self, *args): """ Exits the cluster. This shuts down all existing session connections. """ self.cluster.__exit__(*args) def add_movies(self, table_name, movie_file_path): """ Gets movies from a JSON file and adds them to a table in the keyspace. :param table_name: The name of the table. :param movie_file_path: The path and file name of a JSON file that contains movie data. """ with open(movie_file_path, "r") as movie_file: movies = json.loads(movie_file.read()) stmt = self.session.prepare( f"INSERT INTO {table_name} (year, title, release_date, plot) VALUES (?, ?, ?, ?);" ) for movie in movies[:20]: self.session.execute( stmt, parameters=[ movie["year"], movie["title"], date.fromisoformat(movie["info"]["release_date"].partition("T")[0]), movie["info"]["plot"], ], ) def get_movies(self, table_name, watched=None): """ Gets the title and year of the full list of movies from the table. :param table_name: The name of the movie table. :param watched: When specified, the returned list of movies is filtered to either movies that have been watched or movies that have not been watched. Otherwise, all movies are returned. :return: A list of movies in the table. """ if watched is None: stmt = SimpleStatement(f"SELECT title, year from {table_name}") params = None else: stmt = SimpleStatement( f"SELECT title, year from {table_name} WHERE watched = %s ALLOW FILTERING" ) params = [watched] return self.session.execute(stmt, parameters=params).all() def get_movie(self, table_name, title, year): """ Gets a single movie from the table, by title and year. :param table_name: The name of the movie table. :param title: The title of the movie. :param year: The year of the movie's release. :return: The requested movie. """ return self.session.execute( SimpleStatement( f"SELECT * from {table_name} WHERE title = %s AND year = %s" ), parameters=[title, year], ).one() def watched_movie(self, table_name, title, year): """ Updates a movie as having been watched. :param table_name: The name of the movie table. :param title: The title of the movie. :param year: The year of the movie's release. """ self.session.execute( SimpleStatement( f"UPDATE {table_name} SET watched=true WHERE title = %s AND year = %s" ), parameters=[title, year], )
-
API の詳細については、AWS SDK for Python (Boto3) API リファレンスの以下のトピックを参照してください。
-