利用用户定义的函数 (UDF)

创建 Amazon CloudFormation 堆栈。

1. 转至 Amazon Web Services Management Console 然后在服务列表下选择 CloudFormation。

2. 在 CloudFormation 页面，选择 堆栈，然后选择以新资源创建堆栈（标准）。

3. 在创建堆栈页面上，选择上传模板文件，然后选择您之前下载的 kda-flink-udf.yml。上传文件，然后选择 下一步。

4. 为模板指定一个名称 (比如 kinesis-UDF)，以便记忆。如果您想更改名称，请更新输入参数，例如输入流。选择 Next（下一步）。

5. 在配置堆栈选项页面上，根据需要添加标签，然后选择下一步。

6. 在查看页面上，选中允许创建 IAM 资源的复选框，然后选择提交。

使用笔记本

1. 转至 Amazon Web Services Management Console 然后在服务列表下选择 Amazon Kinesis。

2. 在左侧导航页面上，选择 Analytics 应用程序，然后选择 Studio 笔记本。

3. 验证 KinesisDataAnalyticsStudio 笔记本是否正在运行。

4. 选择笔记本，然后选择在 Apache Zeppelin 中打开。

5. 下载 数据创建器 Zeppelin 笔记本文件，您需要使用该文件读取数据并将数据加载到 Kinesis 流中。

6. 导入 Data Producer Zeppelin 笔记本。请务必修改笔记本代码中的输入 STREAM_NAME 和 REGION。输入流名称可以在Amazon CloudFormation堆栈输出中找到。

7. 选择运行此段落按钮，将示例数据插入输入的 Kinesis 数据流，执行数据创建器笔记本。

8. 示例数据加载过程中，下载 MaskPhoneNumber 交互式笔记本，它将读取输入数据，匿名化输入流中的电话号码，并将匿名化数据存储到输出流中。

9. 导入 MaskPhoneNumber-interactive Zeppelin 笔记本。

10. 执行笔记本中的每个段落。

    1. 在第 1 段中，通过导入用户定义函数来匿名化电话号码。

       %flink(parallelism=1)
       import com.mycompany.app.MaskPhoneNumber
       stenv.registerFunction("MaskPhoneNumber", new MaskPhoneNumber())

    2. 在下一段中，通过创建一个内存表来读取输入流数据。确保流名称和 Amazon 区域正确无误。

       %flink.ssql(type=update)
       
       DROP TABLE IF EXISTS customer_reviews;
       
       CREATE TABLE customer_reviews (
       customer_id VARCHAR,
       product VARCHAR,
       review VARCHAR,
       phone VARCHAR
       )
       WITH (
       'connector' = 'kinesis',
       'stream' = 'KinesisUDFSampleInputStream',
       'aws.region' = 'us-east-1',
       'scan.stream.initpos' = 'LATEST',
       'format' = 'json');

    3. 检查数据是否已加载到内存表中。

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

    4. 调用用户定义的函数对电话号码进行匿名化处理。

       %flink.ssql(type=update)
       select customer_id, product, review, MaskPhoneNumber('mask_phone', phone) as phoneNumber from customer_reviews

    5. 电话号码屏蔽后，创建一个屏蔽号码的视图。

       %flink.ssql(type=update)
       
       DROP VIEW IF EXISTS sentiments_view;
       
       CREATE VIEW  
           sentiments_view
       AS
         select customer_id, product, review, MaskPhoneNumber('mask_phone', phone) as phoneNumber from customer_reviews

    6. 验证数据。

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

    7. 为输出的 Kinesis 流创建内存表。确保流名称和 Amazon 区域正确无误。

       %flink.ssql(type=update)
       
       DROP TABLE IF EXISTS customer_reviews_stream_table;
       
       CREATE TABLE customer_reviews_stream_table (
       customer_id VARCHAR,
       product VARCHAR,
       review VARCHAR,
       phoneNumber varchar 
       )
       WITH (
       'connector' = 'kinesis',
       'stream' = 'KinesisUDFSampleOutputStream',
       'aws.region' = 'us-east-1',
       'scan.stream.initpos' = 'TRIM_HORIZON',
       'format' = 'json');
       

    8. 在目标 Kinesis 流中插入更新的记录。

       %flink.ssql(type=update)
       INSERT INTO customer_reviews_stream_table
       SELECT customer_id, product, review, phoneNumber
       FROM sentiments_view
       

    9. 查看和验证来自目标 Kinesis 流的数据。

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

1. 在 Amazon Web Services Management Console，选择您的笔记本，然后在部署为应用程序配置 (可选) 中，选择编辑。

2. 在 Amazon S3 中代码目标项下，选择通过Amazon CloudFormation 脚本创建的 Amazon S3 存储桶。此过程可能耗时数分钟。

3. 您将无法按原样发布笔记。如果尝试按原样发布，您将遇到不支持 Select 语句的报错。要避免出现此问题，请下载 MaskPhoneNumber 流式处理Zeppelin 笔记本。

4. 导入 MaskPhoneNumber-streaming Zeppelin 笔记本。

5. 打开笔记，然后选择 KinesisDataAnalyticsStudio 操作。

6. 选择构建 maskPhoneNumber 流式处理并导出到 S3。确保重命名应用程序名称，请勿使用任何特殊字符。

7. 选择构建并导出。设置流应用程序将花费数分钟。

8. 构建完成后，选择使用 Amazon 控制台部署。

9. 在下一页，查看设置并确保选择正确的 IAM 角色。接下来，选择创建流应用程序。

10. 几分钟后，您将看到一条消息，提示流应用程序已成功创建。