使用 Kinesis Data Streams 创建创建 Guo

创建表

1. 登录 Amazon Web Services Management Console，然后打开 Amazon Glue 控制台，网址为：https://console.aws.amazon.com/glue/。

2. 如果您还没有Amazon Glue数据库，请从左侧导航栏中选择数据库。选择 “添加数据库”。在添加数据库窗口中default，输入数据库名称。选择创建。

3. 在左侧导航栏中，选择表。在表格页面中，选择添加表，手动添加表。

4. 在设置表的属性页面中stock，输入表名。确保选择之前创建的数据库。选择下一步。

5. 在添加数据存储页面中，选择 Kinesis。对于直播名称，输入ExampleInputStream。对于 Kinesis 源网址，选择回车键https://kinesis.us-east-1.amazonaws.com。如果您复制并粘贴 Kinesis 源网址，请务必删除所有开头或结尾的空格。选择下一步。

6. 在分类页面中，选择 JSON。选择下一步。

7. 在 “定义架构” 页面中，选择 “添加列” 以添加列。添加具有以下属性的列：

   列名称	数据类型
   自动收报机	字符串
   价格	double

   选择下一步。

8. 在下一页上，验证您的设置，然后选择完成。

9. 从表列表中选择新创建的表。

10. 选择编辑表并添加带有键kinesisanalytics.proctime和值的属性proctime。

11. 选择 Apply（应用）。

1. 打开 Kinesis Data Analytics 控制台，网址为 https://console.aws.amazon.com/kinesisanalytics/home?region=us-east-1#/applications/dashboard。

2. 在 Kinesis Data Analytics 应用程序页面中，选择 Studio 选项卡。选择创建 Studio 笔记本。

   注意

   您还可以从亚马逊 MSK 或 Kinesis Data Streams 控制台创建 Studio 笔记本，方法是选择您的输入亚马逊 MSK 集群或 Kinesis 数据流，然后选择 “实时处理数据”。

3. 在创建创建 Studio 笔记本页面上，提供以下信息

   • 输入笔记本MyNotebook的名称。

   • 为 AmazonGlue 数据库选择默认值。

   选择创建 Studio 笔记本。

4. 在MyNotebook页面中，选择 “运行”。等待状态显示正在运行。笔记本电脑运行时收费。

1. 验证账户 ID。您需要此值才能创建应用程序。

2. 创建角色arn:aws:iam::AccountID:role/ZeppelinRole并将以下权限添加到控制台自动创建的角色中。

   "kinesis:GetShardIterator",

   "kinesis:GetRecords",

   "kinesis:ListShards"

3. 创建以下内容的名为 create.json 的文件。用您的信息替换占位符值。

   {
       "ApplicationName": "MyNotebook",
       "RuntimeEnvironment": "ZEPPELIN-FLINK-2_0",
       "ApplicationMode": "INTERACTIVE",
       "ServiceExecutionRole": "arn:aws:iam::AccountID:role/ZeppelinRole",
       "ApplicationConfiguration": {
           "ApplicationSnapshotConfiguration": {
               "SnapshotsEnabled": false
           },
           "ZeppelinApplicationConfiguration": {
               "CatalogConfiguration": {
                   "GlueDataCatalogConfiguration": {
                       "DatabaseARN": "arn:aws:glue:us-east-1:AccountID:database/default"
                   }
               }
           }
       }
   }
   

4. 运行以下命令创建应用程序。

   aws kinesisanalyticsv2 create-application --cli-input-json file://create.json 
   

5. 命令完成后，您将看到显示新 Studio 笔记本电脑详细信息的输出。下面是输出的一个示例。

   {
       "ApplicationDetail": {
           "ApplicationARN": "arn:aws:kinesisanalytics:us-east-1:012345678901:application/MyNotebook",
           "ApplicationName": "MyNotebook",
           "RuntimeEnvironment": "ZEPPELIN-FLINK-2_0",
           "ApplicationMode": "INTERACTIVE",
           "ServiceExecutionRole": "arn:aws:iam::012345678901:role/ZeppelinRole",
   ...
   

6. 运行以下命令以启动应用程序。将示例值替换为您的账户 ID。

   aws kinesisanalyticsv2 start-application --application-arn arn:aws:kinesisanalytics:us-east-1:012345678901:application/MyNotebook\
   

1. 打开 Kinesis 数据生成器。

2. 选择 “使用” 创建 Cognito 用户 CloudFormation。

3. Amazon CloudFormation控制台以 Kinesis 数据生成器模板打开。选择下一步。

4. 在指定堆栈详细信息页面中，输入您的 Cognito 用户的用户名和密码。选择下一步。

5. 在配置堆栈选项页面中，选择下一步。

6. 在 “查看 Kinesis-Data-Generator-Cognito-User” 页面中，选择Amazon CloudFormation 可能创建 IAM 资源的 “我确认”。 复选框。选择Create Stack（创建堆栈）。

7. 等待Amazon CloudFormation堆栈完成创建。堆栈完成后，在Amazon CloudFormation控制台中打开 Kinesis-Data-Generator-Cognito-User 堆栈，然后选择 “输出” 选项卡。打开列出的KinesisDataGeneratorUrl输出值的 URL。

8. 在 Amazon Kinesis 数据生成器页面中，使用您在步骤 4 中创建的证书登录。

9. 在下一页上，提供以下值

   区域	us-east-1
   传输流/传输流	ExampleInputStream
   每秒记录数	1

   对于 “记录模板”，粘贴以下代码：

   {
       "ticker": "{{random.arrayElement(
           ["AMZN","MSFT","GOOG"]
       )}}",
       "price": {{random.number(
           {
               "min":10,
               "max":150
           }
       )}}
   }
   

10. 选择 “发送数据”。

11. 生成器会将数据发送到 Kinesis 数据流。

    在完成下一节的同时，让生成器保持运行状态。

1. 打开 Kinesis Data Analytics 控制台，网址为 https://console.aws.amazon.com/kinesisanalytics/home?region=us-east-1#/applications/dashboard。

2. 在 Kinesis Data Analytics 应用程序页面上，选择 Studio 笔记本选项卡。选择MyNotebook。

3. 在MyNotebook页面中，选择 “在 Apache Zeppelin 中打开”。

   Apache Zeppelin 接口在新选项卡中打开。

4. 在《欢迎来到齐柏林》中！ 页面，选择 “齐柏林笔记”。

5. 在 Zeppelin Note 页面中，在新笔记中输入以下查询：

   %flink.ssql(type=update)
   select * from stock

   选择运行图标。

   片刻之后，笔记会显示来自 Kinesis 数据流的数据。