Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅中国的 Amazon Web Services 服务入门。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

例如：从不同账户的 Kinesis Street 中读取

该示例说明了如何创建 Amazon Kinesis Data Analytics 应用程序，以从不同账户的 Kinesis 流中读取数据。在本示例中，您将一个账户用于源 Kinesis 流，并将第二个账户用于 Kinesis Data Analytics 应用程序和接收器 Kinesis 流。

先决条件

设置

您将使用命名的配置文件访问两个 Amazon 账户。修改Amazon凭证和配置文件中包含两个配置文件，其中包含两个账户的区域和连接信息。

以下示例凭证文件包含两个命名的配置文件：ka-source-stream-account-profile 和 ka-sink-stream-account-profile。将您用于入门教程的账户作为接收器流账户。

[ka-source-stream-account-profile]
aws_access_key_id=AKIAIOSFODNN7EXAMPLE
aws_secret_access_key=wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY

[ka-sink-stream-account-profile]
aws_access_key_id=AKIAI44QH8DHBEXAMPLE
aws_secret_access_key=je7MtGbClwBF/2Zp9Utk/h3yCo8nvbEXAMPLEKEY

以下示例配置文件包含具有区域和输出格式信息的相同命名配置文件。

[profile ka-source-stream-account-profile]
region=us-west-2
output=json

[profile ka-sink-stream-account-profile]
region=us-west-2
output=json

有关将命名配置文件与Amazon CLI请参阅,命名配置文件中的Amazon Command Line Interface文档中）。

创建源 Kinesis 流

在本节中，您在源账户中创建 Kinesis 流。

输入以下命令以创建 Kinesis Streams，应用程序将该流用于输入。请注意，--profile 参数指定要使用的账户配置文件。

$ aws kinesis create-stream \
--stream-name SourceAccountExampleInputStream \
--shard-count 1 \
--profile ka-source-stream-account-profile

创建和更新 IAM 角色和策略

允许跨越对象访问Amazon账户中，您可以在源账户中创建 IAM 角色和策略。然后，您在接收器账户中修改 IAM 策略。有关创建 IAM 角色和策略的信息，请参阅中的以下主题：Amazon Identity and Access Management用户指南：

接收器账户角色和策略

源账户角色和策略

更新 Python 脚本

在本节中，您更新生成示例数据的 Python 脚本以使用源账户配置文件。

使用以下突出显示的更改更新 stock.py 脚本。

import json
import boto3
import random
import datetime
import os

os.environ['AWS_PROFILE'] ='ka-source-stream-account-profile'
os.environ['AWS_DEFAULT_REGION'] = 'us-west-2'


kinesis = boto3.client('kinesis')
def getReferrer():
    data = {}
    now = datetime.datetime.now()
    str_now = now.isoformat()
    data['EVENT_TIME'] = str_now
    data['ticker'] = random.choice(['AAPL', 'AMZN', 'MSFT', 'INTC', 'TBV'])
    price = random.random() * 100
    data['price'] = round(price, 2)
    return data

while True:
        data = json.dumps(getReferrer())
        print(data)
        kinesis.put_record(
                StreamName="SourceAccountExampleInputStream",
                Data=data,
                PartitionKey="partitionkey")

更新 Java 应用程序

在本节中，您更新 Java 应用程序代码，以便从源流中读取时担任源账户角色。

对 BasicStreamingJob.java 文件进行以下更改。将示例源账号 (SOURCE01234567) 替换为您的源账号。

package com.amazonaws.services.kinesisanalytics;

import com.amazonaws.services.kinesisanalytics.runtime.KinesisAnalyticsRuntime;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer;
import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisProducer;
import org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants;
import org.apache.flink.streaming.connectors.kinesis.config.AWSConfigConstants;

import java.io.IOException;
import java.util.Map;
import java.util.Properties;

 /**
 * A basic Kinesis Data Analytics for Java application with Kinesis data streams
 * as source and sink.
 */
public class BasicStreamingJob {
    private static final String region = "us-west-2";
    private static final String inputStreamName = "SourceAccountExampleInputStream";
    private static final String outputStreamName = ExampleOutputStream;
    private static final String roleArn = "arn:aws:iam::SOURCE01234567:role/KA-Source-Stream-Role";
    private static final String roleSessionName = "ksassumedrolesession";

    private static DataStream<String> createSourceFromStaticConfig(StreamExecutionEnvironment env) {
        Properties inputProperties = new Properties();
        inputProperties.setProperty(AWSConfigConstants.AWS_CREDENTIALS_PROVIDER, "ASSUME_ROLE");
        inputProperties.setProperty(AWSConfigConstants.AWS_ROLE_ARN, roleArn);
        inputProperties.setProperty(AWSConfigConstants.AWS_ROLE_SESSION_NAME, roleSessionName);
        inputProperties.setProperty(ConsumerConfigConstants.AWS_REGION, region);
        inputProperties.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "LATEST");

        return env.addSource(new FlinkKinesisConsumer<>(inputStreamName, new SimpleStringSchema(), inputProperties));
    }

    private static FlinkKinesisProducer<String> createSinkFromStaticConfig() {
        Properties outputProperties = new Properties();
        outputProperties.setProperty(ConsumerConfigConstants.AWS_REGION, region);
        outputProperties.setProperty("AggregationEnabled", "false");

        FlinkKinesisProducer<String> sink = new FlinkKinesisProducer<>(new SimpleStringSchema(), outputProperties);
        sink.setDefaultStream(outputStreamName);
        sink.setDefaultPartition("0");
        return sink;
    }

    public static void main(String[] args) throws Exception {
        // set up the streaming execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> input = createSourceFromStaticConfig(env);

        input.addSink(createSinkFromStaticConfig());

        env.execute("Flink Streaming Java API Skeleton");
    }
}

构建、上传和运行应用程序

执行以下操作以更新和运行应用程序：

现在，应用程序从另一个账户的 Kinesis 流中读取数据。

您可以检查 ExampleOutputStream 流的 PutRecords.Bytes 指标，以验证应用程序是否正常工作。如果在输出流中具有活动，则应用程序正常工作。