Apache Flink용 관리 서비스 응용 프로그램을 만들고 실행하십시오. - Managed Service for Apache Flink

Amazon Managed Service for Apache Flink는 이전에 Amazon Kinesis Data Analytics for Apache Flink로 알려졌습니다.

기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

Apache Flink용 관리 서비스 응용 프로그램을 만들고 실행하십시오.

이 단계에서는 Kinesis 데이터 스트림을 소스로 사용하고 싱크를 사용하여 Apache Flink용 관리 서비스 애플리케이션을 생성합니다.

종속 리소스 생성

이 연습을 위해 Managed Service for Apache Flink 애플리케이션을 생성하기 전에 다음과 같은 종속 리소스를 생성해야 합니다.

  • 입력 및 출력용 Kinesis 데이터 스트림 2개

  • 애플리케이션 코드를 저장하는 Amazon S3 버킷

    참고

    이 자습서에서는 us-east-1 미국 동부 (버지니아 북부) 지역에 애플리케이션을 배포한다고 가정합니다. 다른 지역을 사용하는 경우 모든 단계를 적절히 조정하십시오.

두 개의 Amazon Kinesis 데이터 스트림을 생성합니다.

이 연습을 위해 Managed Service for Apache Flink 애플리케이션을 생성하기 전에 두 개의 Kinesis 데이터 스트림(ExampleInputStreamExampleOutputStream)을 생성하세요. 이 애플리케이션은 애플리케이션 소스 및 대상 스트림에 대해 이러한 스트림을 사용합니다.

Amazon Kinesis 콘솔 또는 다음 AWS CLI 명령을 사용하여 이러한 스트림을 생성할 수 있습니다. 콘솔 지침은 Amazon Kinesis Data Streams 개발자 가이드데이터 스트림 생성 및 업데이트를 참조하세요. 를 사용하여 스트림을 AWS CLI생성하려면 다음 명령을 사용하여 애플리케이션에 사용하는 지역에 맞게 조정하십시오.

데이터 스트림 (AWS CLI)을 생성하려면
  1. 첫 번째 스트림 (ExampleInputStream) 을 생성하려면 다음 Amazon Kinesis 명령을 create-stream AWS CLI 사용하십시오.

    $ aws kinesis create-stream \ --stream-name ExampleInputStream \ --shard-count 1 \ --region us-east-1 \
  2. 애플리케이션이 출력을 작성하는 데 사용하는 두 번째 스트림을 생성하려면 동일한 명령을 실행하여 스트림 이름을 다음과 같이 변경합니다. ExampleOutputStream

    $ aws kinesis create-stream \ --stream-name ExampleOutputStream \ --shard-count 1 \ --region us-east-1 \

애플리케이션 코드용 Amazon S3 버킷 생성

콘솔을 사용하여 Amazon S3 버킷을 생성할 수 있습니다. 콘솔을 사용하여 Amazon S3 버킷을 생성하는 방법을 알아보려면 Amazon S3 사용 설명서의 버킷 생성을 참조하십시오. 전역적으로 고유한 이름을 사용하여 Amazon S3 버킷의 이름을 지정합니다 (예: 로그인 이름 추가).

참고

이 자습서에서 사용하는 리전 (us-east-1) 에 버킷을 생성해야 합니다.

기타 리소스

애플리케이션을 생성할 때 Apache Flink용 관리형 서비스는 다음과 같은 Amazon CloudWatch 리소스를 자동으로 생성합니다 (아직 존재하지 않는 경우).

  • /AWS/KinesisAnalytics-java/<my-application>라는 로그 그룹.

  • kinesis-analytics-log-stream라는 로그 스트림.

로컬 개발 환경 설정

개발 및 디버깅을 위해 원하는 시스템에서 직접 Apache Flink 애플리케이션을 실행할 수 있습니다. IDE 모든 아파치 플링크 종속성은 Apache Maven을 사용하여 일반 자바 종속성처럼 처리됩니다.

참고

개발 시스템에는 자바 JDK 11, Maven 및 Git이 설치되어 있어야 합니다. 이클립스, 자바, 네온 또는 IntelliJ와 같은 개발 환경을 사용하는 것이 좋습니다. IDEA 모든 사전 요구 사항을 충족하는지 확인하려면 을 참조하십시오. 연습을 완료하기 위한 사전 요구 사항을 충족하십시오. 컴퓨터에 Apache Flink 클러스터를 설치할 필요는 없습니다.

세션을 인증하세요. AWS

애플리케이션은 Kinesis 데이터 스트림을 사용하여 데이터를 게시합니다. 로컬에서 실행하는 경우 Kinesis 데이터 스트림에 쓸 수 있는 권한이 있는 유효한 AWS 인증 세션이 있어야 합니다. 다음 단계를 사용하여 세션을 인증하십시오.

  1. 유효한 자격 증명이 구성된 AWS CLI 및 이름이 지정된 프로필이 없는 경우 을 참조하십시오. AWS Command Line Interface (AWS CLI) 를 설정합니다.

  2. 다음 테스트 기록을 게시하여 AWS CLI 가 올바르게 구성되어 있고 사용자에게 Kinesis 데이터 스트림에 쓸 수 있는 권한이 있는지 확인하십시오.

    $ aws kinesis put-record --stream-name ExampleOutputStream --data TEST --partition-key TEST
  3. 통합할 IDE 플러그인이 있는 AWS경우 이를 사용하여 에서 실행 중인 애플리케이션에 자격 증명을 전달할 수 있습니다. IDE 자세한 내용은 IDEAIntelliJ용AWS 툴킷 AWS 및 Eclipse용 툴킷을 참조하십시오.

Apache Flink 스트리밍 Java 코드를 다운로드하여 검토하십시오.

이 예제의 Java 애플리케이션 코드는 에서 제공됩니다. GitHub 애플리케이션 코드를 다운로드하려면 다음을 수행하세요.

  1. 다음 명령을 사용하여 원격 리포지토리를 복제합니다.

    git clone https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples.git
  2. amazon-managed-service-for-apache-flink-examples/tree/main/java/GettingStarted 디렉터리로 이동합니다.

애플리케이션 구성 요소 검토

애플리케이션은 전적으로 com.amazonaws.services.msf.BasicStreamingJob 클래스에서 구현됩니다. 이 main() 메서드는 스트리밍 데이터를 처리하고 실행하기 위한 데이터 흐름을 정의합니다.

참고

최적화된 개발자 경험을 위해 애플리케이션은 Apache Flink용 Amazon Managed Service와 로컬에서 코드 변경 없이 실행되도록 설계되어 사용자 환경에서 개발할 수 있습니다. IDE

  • Amazon Managed Service for Apache Flink 및 사용자 IDE 환경에서 실행될 때 작동하도록 런타임 구성을 읽으려면 애플리케이션이 로컬에서 독립 실행형으로 실행되고 있는지 자동으로 감지합니다. IDE 이 경우 애플리케이션은 런타임 구성을 다르게 로드합니다.

    1. 응용 프로그램이 에서 독립 실행형 모드로 실행 중인 것을 감지하면 프로젝트의 resources 폴더에 포함된 application_properties.json 파일을 구성하십시오IDE. 파일 내용은 다음과 같습니다.

    2. 애플리케이션이 Apache Flink용 Amazon Managed Service에서 실행되는 경우 기본 동작은 Apache Flink용 Amazon Managed Service 애플리케이션에 정의한 런타임 속성에서 애플리케이션 구성을 로드합니다. Apache Flink용 관리 서비스 애플리케이션 생성 및 구성을 참조하세요.

      private static Map<String, Properties> loadApplicationProperties(StreamExecutionEnvironment env) throws IOException { if (env instanceof LocalStreamEnvironment) { LOGGER.info("Loading application properties from '{}'", LOCAL_APPLICATION_PROPERTIES_RESOURCE); return KinesisAnalyticsRuntime.getApplicationProperties( BasicStreamingJob.class.getClassLoader() .getResource(LOCAL_APPLICATION_PROPERTIES_RESOURCE).getPath()); } else { LOGGER.info("Loading application properties from Amazon Managed Service for Apache Flink"); return KinesisAnalyticsRuntime.getApplicationProperties(); } }
  • main() 메서드는 애플리케이션 데이터 흐름을 정의하고 실행합니다.

    • 기본 스트리밍 환경을 초기화합니다. 이 예제에서는 와 함께 사용할 파일과 StreamExecutionEnvironment 함께 사용할 데이터 DataSteam API 및 테이블을 만드는 방법을 보여 SQL 줍니다. StreamTableEnvironment API 두 환경 객체는 동일한 런타임 환경에 대한 두 개의 개별 참조이므로 다르게 사용할 수 APIs 있습니다.

      StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    • 애플리케이션 구성 매개변수를 로드합니다. 그러면 애플리케이션이 실행되는 위치에 따라 올바른 위치에서 자동으로 로드됩니다.

      Map<String, Properties> applicationParameters = loadApplicationProperties(env);
    • 애플리케이션은 Kinesis Consumer 커넥터를 사용하여 입력 스트림에서 데이터를 읽는 소스를 정의합니다. 입력 스트림의 구성은 PropertyGroupId InputStream0 =에 정의되어 있습니다. 스트림의 이름과 지역은 aws.region 각각 stream.name 및 라는 속성에 있습니다. 단순화를 위해 이 소스는 레코드를 문자열로 읽습니다.

      private static FlinkKinesisConsumer<String> createSource(Properties inputProperties) { String inputStreamName = inputProperties.getProperty("stream.name"); return new FlinkKinesisConsumer<>(inputStreamName, new SimpleStringSchema(), inputProperties); } ... public static void main(String[] args) throws Exception { ... SourceFunction<String> source = createSource(applicationParameters.get("InputStream0")); DataStream<String> input = env.addSource(source, "Kinesis Source"); ... }
    • 그런 다음 애플리케이션은 Kinesis Streams Sink 커넥터를 사용하여 싱크를 정의하여 출력 스트림으로 데이터를 전송합니다. 출력 스트림 이름과 지역은 입력 스트림과 OutputStream0 마찬가지로 PropertyGroupId =에 정의됩니다. 싱크는 소스에서 데이터를 DataStream 가져오는 내부에 직접 연결됩니다. 실제 애플리케이션에서는 소스와 싱크 간에 약간의 변형이 있습니다.

      private static KinesisStreamsSink<String> createSink(Properties outputProperties) { String outputStreamName = outputProperties.getProperty("stream.name"); return KinesisStreamsSink.<String>builder() .setKinesisClientProperties(outputProperties) .setSerializationSchema(new SimpleStringSchema()) .setStreamName(outputStreamName) .setPartitionKeyGenerator(element -> String.valueOf(element.hashCode())) .build(); } ... public static void main(String[] args) throws Exception { ... Sink<String> sink = createSink(applicationParameters.get("OutputStream0")); input.sinkTo(sink); ... }
    • 마지막으로 방금 정의한 데이터 흐름을 실행합니다. 이것은 데이터 흐름에 필요한 모든 연산자를 정의한 후의 main() 메서드의 마지막 명령이어야 합니다.

      env.execute("Flink streaming Java API skeleton");

pom.xml 파일 사용

pom.xml 파일은 애플리케이션에 필요한 모든 종속성을 정의하고 Flink에 필요한 모든 종속성을 포함하는 fat-jar를 빌드하도록 Maven Shade 플러그인을 설정합니다.

  • provided일부 종속성에는 범위가 있습니다. 이러한 종속성은 애플리케이션이 Apache Flink용 Amazon Managed Service에서 실행될 때 자동으로 사용할 수 있습니다. 애플리케이션을 컴파일하거나 애플리케이션을 로컬에서 실행하는 데 필요합니다. IDE 자세한 내용은 애플리케이션을 로컬에서 실행합니다. 단원을 참조하십시오. Apache Flink용 Amazon 관리형 서비스에서 사용할 런타임과 동일한 Flink 버전을 사용하고 있는지 확인하십시오.

    <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency>
  • 이 애플리케이션에서 사용하는 Kinesis 커넥터와 같은 기본 범위를 사용하여 pom에 Apache Flink 종속성을 추가해야 합니다. 자세한 내용은 아파치 플링크용 매니지드 서비스와 함께 아파치 플링크 커넥터 사용하기 단원을 참조하십시오. 애플리케이션에 필요한 추가 Java 종속성을 추가할 수도 있습니다.

    <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kinesis</artifactId> <version>${aws.connector.version}</version> </dependency>
  • Maven Java Compiler 플러그인을 사용하면 현재 아파치 플링크에서 지원하는 JDK 버전인 Java 11을 기준으로 코드를 컴파일할 수 있습니다.

  • Maven Shade 플러그인은 런타임에서 제공하는 일부 라이브러리를 제외하고 fat-jar를 패키징합니다. 또한 두 개의 변환기를 지정합니다: 및. ServicesResourceTransformer ManifestResourceTransformer 후자는 응용 프로그램을 시작하는 main 메서드가 포함된 클래스를 구성합니다. 메인 클래스의 이름을 바꾸는 경우 이 트랜스포머를 업데이트하는 것을 잊지 마세요.

  • <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> ... <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> <mainClass>com.amazonaws.services.msf.BasicStreamingJob</mainClass> </transformer> ... </plugin>

입력 스트림에 샘플 레코드를 작성합니다.

이 섹션에서는 애플리케이션이 처리할 수 있도록 샘플 레코드를 스트림으로 전송합니다. Python 스크립트 또는 Kinesis 데이터 생성기를 사용하여 샘플 데이터를 생성하는 두 가지 옵션이 있습니다.

Python 스크립트를 사용하여 샘플 데이터 생성

Python 스크립트를 사용하여 샘플 레코드를 스트림으로 보낼 수 있습니다.

참고

이 Python 스크립트를 실행하려면 Python 3.x를 사용하고 AWS SDKPython용 (Boto) 라이브러리가 설치되어 있어야 합니다.

Kinesis 입력 스트림으로 테스트 데이터 전송을 시작하려면:

  1. 데이터 생성기 GitHub 리포지토리에서 데이터 생성기 stock.py Python 스크립트를 다운로드합니다.

  2. stock.py 스크립트를 실행합니다.

    $ python stock.py

자습서의 나머지 부분을 완료하는 동안 스크립트를 계속 실행하세요. 이제 Apache Flink 애플리케이션을 실행할 수 있습니다.

Kinesis 데이터 생성기를 사용하여 샘플 데이터 생성

Python 스크립트를 사용하는 대신 호스팅 버전으로도 제공되는 Kinesis Data Generator를 사용하여 무작위 샘플 데이터를 스트림에 보낼 수 있습니다. Kinesis Data Generator는 브라우저에서 실행되므로 시스템에 아무것도 설치할 필요가 없습니다.

Kinesis 데이터 생성기를 설정하고 실행하려면

  1. Kinesis 데이터 생성기 설명서의 지침에 따라 도구에 대한 액세스 권한을 설정합니다. 사용자 및 암호를 설정하는 AWS CloudFormation 템플릿을 실행합니다.

  2. 템플릿에서 생성한 데이터를 통해 Kinesis 데이터 URL 생성기에 액세스할 수 있습니다 CloudFormation. CloudFormation 템플릿이 URL 완성된 후 출력 탭에서 찾을 수 있습니다.

  3. 데이터 생성기 구성:

    • 지역: 이 자습서에서 사용할 지역을 선택합니다. us-east-1

    • 스트림/전송 스트림: 애플리케이션에서 사용할 입력 스트림을 선택합니다. ExampleInputStream

    • 초당 레코드 수: 100

    • 레코드 템플릿: 다음 템플릿을 복사하여 붙여넣습니다.

      { "event_time" : "{{date.now("YYYY-MM-DDTkk:mm:ss.SSSSS")}}, "ticker" : "{{random.arrayElement( ["AAPL", "AMZN", "MSFT", "INTC", "TBV"] )}}", "price" : {{random.number(100)}} }
  4. 템플릿 테스트: 테스트 템플릿을 선택하고 생성된 레코드가 다음과 유사한지 확인합니다.

    { "event_time" : "2024-06-12T15:08:32.04800, "ticker" : "INTC", "price" : 7 }
  5. 데이터 생성기 시작: 데이터 전송 선택을 선택합니다.

Kinesis 데이터 생성기가 이제 에 데이터를 보내고 있습니다. ExampleInputStream

애플리케이션을 로컬에서 실행합니다.

에서 로컬로 Flink 애플리케이션을 실행하고 디버그할 수 있습니다. IDE

참고

계속하기 전에 입력 및 출력 스트림을 사용할 수 있는지 확인하세요. 두 개의 Amazon Kinesis 데이터 스트림을 생성합니다.을 참조하세요. 또한 두 스트림 모두에서 읽고 쓸 수 있는 권한이 있는지 확인하십시오. 세션을 인증하세요. AWS을 참조하세요.

로컬 개발 환경을 설정하려면 Java 11JDK, Apache Maven 및 IDE Java 개발이 필요합니다. 필수 사전 요구 사항을 충족하는지 확인하십시오. 연습을 완료하기 위한 사전 요구 사항을 충족하십시오.을 참조하세요.

Java 프로젝트를 사용자 컴퓨터로 가져옵니다. IDE

에서 애플리케이션 작업을 시작하려면 Java 프로젝트로 가져와야 합니다. IDE

클론한 리포지토리에는 여러 예제가 들어 있습니다. 각 예제는 별도의 프로젝트입니다. 이 자습서에서는 ./java/GettingStarted 하위 디렉터리의 콘텐츠를 사용자 디렉토리로 가져오십시오. IDE

Maven을 사용하여 코드를 기존 Java 프로젝트로 삽입합니다.

참고

새 Java 프로젝트를 가져오는 정확한 프로세스는 사용 IDE 중인 Java 프로젝트에 따라 다릅니다.

로컬 애플리케이션 구성을 확인하세요.

로컬에서 실행하는 경우 애플리케이션은 에 있는 프로젝트의 리소스 폴더에 있는 application_properties.json 파일의 구성을 사용합니다./src/main/resources. 이 파일을 편집하여 다른 Kinesis 스트림 이름 또는 지역을 사용할 수 있습니다.

[ { "PropertyGroupId": "InputStream0", "PropertyMap": { "stream.name": "ExampleInputStream", "flink.stream.initpos": "LATEST", "aws.region": "us-east-1" } }, { "PropertyGroupId": "OutputStream0", "PropertyMap": { "stream.name": "ExampleOutputStream", "aws.region": "us-east-1" } } ]

IDE실행 구성을 설정합니다.

Java 애플리케이션을 실행하는 것처럼 기본 클래스를 com.amazonaws.services.msf.BasicStreamingJob 실행하여 Flink 애플리케이션을 IDE 직접 실행하고 디버그할 수 있습니다. 애플리케이션을 실행하기 전에 Run 구성을 설정해야 합니다. 설정은 사용 IDE 중인 제품에 따라 다릅니다. 예를 들어 IntelliJ 설명서의 실행/디버그 구성을 참조하십시오. IDEA 특히 다음을 설정해야 합니다.

  1. 클래스 경로에 provided 종속성을 추가합니다. 이는 로컬에서 실행할 때 provided 범위가 있는 종속성이 애플리케이션에 전달되도록 하기 위해 필요합니다. 이렇게 설정하지 않으면 응용 프로그램에 class not found 오류가 즉시 표시됩니다.

  2. Kinesis 스트림에 액세스하기 위한 AWS 자격 증명을 애플리케이션에 전달합니다. 가장 빠른 방법은 IDEAIntelliJ용AWS 툴킷을 사용하는 것입니다. 실행 구성에서 이 IDE 플러그인을 사용하여 특정 AWS 프로파일을 선택할 수 있습니다. AWS 인증은 이 프로필을 사용하여 이루어집니다. AWS 자격 증명을 직접 전달할 필요는 없습니다.

  3. JDK11을 사용하여 에서 애플리케이션이 IDE 실행되는지 확인합니다.

에서 애플리케이션을 실행하세요. IDE

에 대한 실행 구성을 설정한 후 일반 Java 응용 프로그램처럼 실행하거나 디버그할 수 있습니다. BasicStreamingJob

참고

Maven에서 생성한 fat-jar는 java -jar ... 명령줄에서 직접 실행할 수 없습니다. 이 jar에는 응용 프로그램을 독립 실행형으로 실행하는 데 필요한 Flink 코어 종속성이 포함되어 있지 않습니다.

응용 프로그램이 성공적으로 시작되면 독립형 미니 클러스터와 커넥터 초기화에 대한 일부 정보가 기록됩니다. 그 다음에는 응용 프로그램이 시작될 때 Flink가 일반적으로 내보내는 여러 INFO WARN 로그와 일부 로그가 이어집니다.

13:43:31,405 INFO com.amazonaws.services.msf.BasicStreamingJob [] - Loading application properties from 'flink-application-properties-dev.json' 13:43:31,549 INFO org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer [] - Flink Kinesis Consumer is going to read the following streams: ExampleInputStream, 13:43:31,676 INFO org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils [] - The configuration option taskmanager.cpu.cores required for local execution is not set, setting it to the maximal possible value. 13:43:31,676 INFO org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils [] - The configuration option taskmanager.memory.task.heap.size required for local execution is not set, setting it to the maximal possible value. 13:43:31,676 INFO org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils [] - The configuration option taskmanager.memory.task.off-heap.size required for local execution is not set, setting it to the maximal possible value. 13:43:31,676 INFO org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils [] - The configuration option taskmanager.memory.network.min required for local execution is not set, setting it to its default value 64 mb. 13:43:31,676 INFO org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils [] - The configuration option taskmanager.memory.network.max required for local execution is not set, setting it to its default value 64 mb. 13:43:31,676 INFO org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils [] - The configuration option taskmanager.memory.managed.size required for local execution is not set, setting it to its default value 128 mb. 13:43:31,677 INFO org.apache.flink.runtime.minicluster.MiniCluster [] - Starting Flink Mini Cluster ....

초기화가 완료되면 애플리케이션은 더 이상 로그 항목을 내보내지 않습니다. 데이터가 흐르는 동안에는 로그가 생성되지 않습니다.

애플리케이션이 데이터를 올바르게 처리하고 있는지 확인하려면 다음 섹션에 설명된 대로 입력 및 출력 Kinesis 스트림을 검사할 수 있습니다.

참고

플로우 데이터에 대한 로그를 내보내지 않는 것은 Flink 애플리케이션의 일반적인 동작입니다. 모든 레코드에서 로그를 내보내는 것은 디버깅에 편리할 수 있지만 프로덕션 환경에서 실행할 때는 상당한 오버헤드가 발생할 수 있습니다.

Kinesis 스트림의 입력 및 출력 데이터 관찰

Amazon Kinesis 콘솔의 데이터 뷰어를 사용하여 (샘플 Python 생성) 또는 Kinesis 데이터 생성기 (링크) 에서 입력 스트림으로 전송된 레코드를 관찰할 수 있습니다.

기록을 관찰하려면
  1. /kinesis에서 Kinesis 콘솔을 엽니다. https://console.aws.amazon.com

  2. 이 자습서를 실행 중인 지역과 동일한지 확인하십시오. 이 지역은 기본적으로 us-east-1 미국 동부 (버지니아 북부) 입니다. 지역이 일치하지 않는 경우 지역을 변경하십시오.

  3. 데이터 스트림을 선택합니다.

  4. 관찰하려는 스트림을 ExampleInputStream 선택하거나 ExampleOutputStream.

  5. 데이터 뷰어 탭을 선택합니다.

  6. 샤드를 선택하고 최신 버전을 시작 위치로 유지한 다음 레코드 가져오기를 선택합니다. “이 요청에 대한 레코드를 찾을 수 없습니다.” 라는 오류가 표시될 수 있습니다. 그렇다면 레코드 가져오기 재시도를 선택하십시오. 스트림에 게시된 최신 레코드가 표시됩니다.

  7. 데이터 열에서 값을 선택하여 레코드 콘텐츠를 JSON 형식별로 검사합니다.

애플리케이션이 로컬에서 실행되지 않도록 하세요.

에서 실행 중인 애플리케이션을 중지하십시오IDE. 는 IDE 일반적으로 “중지” 옵션을 제공합니다. 정확한 위치와 방법은 사용 IDE 중인 제품에 따라 다릅니다.

애플리케이션 코드를 컴파일하고 패키징하세요.

이 섹션에서는 Apache Maven을 사용하여 Java 코드를 컴파일하고 파일로 패키징합니다. JAR Maven 명령줄 도구 또는 사용자 도구를 사용하여 코드를 컴파일하고 패키징할 수 있습니다. IDE

Maven 명령줄을 사용하여 컴파일 및 패키징하려면:

Java GettingStarted 프로젝트가 들어 있는 디렉토리로 이동하고 다음 명령을 실행합니다.

$ mvn package

를 사용하여 컴파일하고 패키징하려면IDE:

IDEMaven mvn package 통합에서 실행하세요.

두 경우 모두 다음 JAR 파일이 생성됩니다. target/amazon-msf-java-stream-app-1.0.jar

참고

에서 “빌드 프로젝트”를 실행하면 JAR 파일이 생성되지 IDE 않을 수 있습니다.

애플리케이션 코드 JAR 파일을 업로드하세요.

이 단원에서는 이전 섹션에서 생성한 JAR 파일을 이 자습서 시작 부분에서 생성한 Amazon Simple Storage Service (Amazon S3) 버킷에 업로드합니다. 이 단계를 완료하지 않은 경우 (링크) 를 참조하십시오.

애플리케이션 코드 JAR 파일을 업로드하려면
  1. 에서 Amazon S3 콘솔을 엽니다 https://console.aws.amazon.com/s3/.

  2. 애플리케이션 코드용으로 이전에 생성한 버킷을 선택합니다.

  3. 업로드를 선택합니다.

  4. [Add Files]를 선택합니다.

  5. 이전 단계에서 생성한 JAR 파일로 이동합니다target/amazon-msf-java-stream-app-1.0.jar.

  6. 다른 설정을 변경하지 않고 업로드를 선택합니다.

주의

에서 올바른 JAR 파일을 선택했는지 확인하십시오<repo-dir>/java/GettingStarted/target/amazon-msf-java-stream-app-1.0.jar.

target디렉터리에는 업로드할 필요가 없는 다른 JAR 파일도 포함되어 있습니다.

Apache Flink용 관리 서비스 애플리케이션 생성 및 구성

콘솔이나 AWS CLI를 사용하여 Managed Service for Apache Flink 애플리케이션을 생성하고 실행할 수 있습니다. 이 자습서에서는 콘솔을 사용하게 됩니다.

참고

콘솔을 사용하여 애플리케이션을 생성하면 AWS Identity and Access Management (IAM) 및 Amazon CloudWatch Logs 리소스가 자동으로 생성됩니다. 를 사용하여 애플리케이션을 생성할 때는 이러한 리소스를 별도로 생성합니다. AWS CLI

애플리케이션 생성

애플리케이션을 생성하려면
  1. /flink에서 아파치 플링크용 관리 서비스 콘솔을 엽니다. https://console.aws.amazon.com

  2. 올바른 지역이 선택되었는지 확인하십시오: us-east-1 미국 동부 (버지니아 북부)

  3. 오른쪽 메뉴를 열고 Apache Flink 애플리케이션을 선택한 다음 스트리밍 애플리케이션 생성을 선택합니다. 또는 초기 페이지의 시작하기 컨테이너에서 스트리밍 애플리케이션 만들기를 선택할 수도 있습니다.

  4. 스트리밍 애플리케이션 생성 페이지에서:

    • 스트림 처리 응용 프로그램을 설정하는 방법을 선택하십시오. 처음부터 만들기를 선택하십시오.

    • 아파치 플링크 구성, 애플리케이션 플링크 버전: 아파치 플링크 1.19를 선택합니다.

  5. 애플리케이션 구성

    • 애플리케이션 이름: 를 입력합니다MyApplication.

    • 설명: 를 입력합니다My java test app.

    • 애플리케이션 리소스 액세스: 필수 정책이 kinesis-analytics-MyApplication-us-east-1 포함된 IAM 역할 생성/업데이트를 선택합니다.

  6. 애플리케이션 설정에 맞게 템플릿을 구성하십시오.

    • 템플릿: 개발을 선택합니다.

  7. 페이지 하단에서 스트리밍 애플리케이션 생성을 선택합니다.

참고

콘솔을 사용하여 Apache Flink용 관리 서비스 애플리케이션을 만들 때 애플리케이션에 대한 IAM 역할 및 정책을 생성할 수 있습니다. 귀하의 애플리케이션은 이 역할 및 정책을 사용하여 종속 리소스에 액세스합니다. 이러한 IAM 리소스의 이름은 다음과 같이 애플리케이션 이름과 지역을 사용하여 지정됩니다.

  • 정책: kinesis-analytics-service-MyApplication-us-east-1

  • 역할: kinesisanalytics-MyApplication-us-east-1

아파치 플링크용 아마존 매니지드 서비스는 이전에 Kinesis Data Analytics로 알려졌습니다. 자동으로 생성되는 리소스 이름에는 이전 버전과의 호환성을 위해 접두사가 붙습니다. kinesis-analytics-

정책 편집 IAM

IAM정책을 편집하여 Kinesis 데이터 스트림에 액세스할 수 있는 권한을 추가합니다.

정책을 편집하려면
  1. 에서 IAM 콘솔을 엽니다 https://console.aws.amazon.com/iam/.

  2. 정책을 선택하세요. 이전 섹션에서 콘솔이 생성한 kinesis-analytics-service-MyApplication-us-east-1 정책을 선택합니다.

  3. 편집을 선택한 다음 JSON탭을 선택합니다.

  4. 다음 정책 예제의 강조 표시된 부분을 정책에 추가하세요. 샘플 계정 교체 IDs (012345678901) 를 계정 ID로 사용하십시오.

    { "Version": "2012-10-17", "Statement": [ { "Sid": "ReadCode", "Effect": "Allow", "Action": [ "s3:GetObject", "s3:GetObjectVersion" ], "Resource": [ "arn:aws:s3:::my-bucket/kinesis-analytics-placeholder-s3-object" ] }, { "Sid": "ListCloudwatchLogGroups", "Effect": "Allow", "Action": [ "logs:DescribeLogGroups" ], "Resource": [ "arn:aws:logs:us-east-1:012345678901:log-group:*" ] }, { "Sid": "ListCloudwatchLogStreams", "Effect": "Allow", "Action": [ "logs:DescribeLogStreams" ], "Resource": [ "arn:aws:logs:us-east-1:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*" ] }, { "Sid": "PutCloudwatchLogs", "Effect": "Allow", "Action": [ "logs:PutLogEvents" ], "Resource": [ "arn:aws:logs:us-east-1:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream" ] }, { "Sid": "ReadInputStream", "Effect": "Allow", "Action": "kinesis:*", "Resource": "arn:aws:kinesis:us-east-1:012345678901:stream/ExampleInputStream" }, { "Sid": "WriteOutputStream", "Effect": "Allow", "Action": "kinesis:*", "Resource": "arn:aws:kinesis:us-east-1:012345678901:stream/ExampleOutputStream" } ] }
  5. 페이지 하단에서 다음을 선택한 후 변경사항 저장을 선택합니다.

애플리케이션 구성

애플리케이션 구성을 편집하여 애플리케이션 코드 아티팩트를 설정합니다.

구성을 편집하려면
  1. MyApplication페이지에서 구성을 선택합니다.

  2. 애플리케이션 코드 위치 섹션에서:

    • Amazon S3 버킷의 경우 애플리케이션 코드용으로 이전에 생성한 버킷을 선택합니다. [Browse] 를 선택하고 올바른 버킷을 선택한 다음 [Choose] 선택합니다. 버킷 이름을 클릭하지 마십시오.

    • Amazon S3 객체 경로에는 amazon-msf-java-stream-app-1.0.jar를 입력합니다.

  3. 액세스 권한에서 필수 정책이 kinesis-analytics-MyApplication-us-east-1 포함된 IAM 역할 생성/업데이트를 선택합니다.

  4. 런타임 속성 섹션에서 다음 속성을 추가합니다.

  5. 새 항목 추가를 선택하고 다음 매개 변수를 각각 추가합니다.

    그룹 ID
    InputStream0 stream.name ExampleInputStream
    InputStream0 aws.region us-east-1
    OutputStream0 stream.name ExampleOutputStream
    OutputStream0 aws.region us-east-1
  6. 다른 섹션은 수정하지 마십시오.

  7. Save changes(변경 사항 저장)를 선택합니다.

참고

Amazon CloudWatch 로깅을 활성화하도록 선택하면 Apache Flink용 관리형 서비스에서 로그 그룹과 로그 스트림을 자동으로 생성합니다. 이러한 리소스의 이름은 다음과 같습니다.

  • 로그 그룹: /aws/kinesis-analytics/MyApplication

  • 로그 스트림: kinesis-analytics-log-stream

애플리케이션을 실행합니다

이제 애플리케이션이 구성되어 실행 준비가 되었습니다.

애플리케이션을 실행하려면
  1. Apache Flink용 Amazon 매니지드 서비스 콘솔에서 [내 애플리케이션] 을 선택하고 [실행] 을 선택합니다.

  2. 다음 페이지인 애플리케이션 복원 구성 페이지에서 최신 스냅샷으로 실행을 선택한 다음 실행을 선택합니다.

    애플리케이션 세부 정보의 상태가 애플리케이션 시작 시점 사이를 Ready Starting Running 오갔다가 다시 시작한 시점으로 바뀝니다.

애플리케이션이 Running 상태에 있으면 이제 Flink 대시보드를 열 수 있습니다.

대시보드 열기
  1. [Apache Flink 대시보드 열기] 를 선택합니다. 대시보드가 새 페이지에서 열립니다.

  2. 실행 중인 작업 목록에서 볼 수 있는 단일 작업을 선택합니다.

    참고

    런타임 속성을 설정하거나 IAM 정책을 잘못 편집하면 애플리케이션 상태가 로 Running 바뀔 수 있지만 Flink 대시보드에는 작업이 계속 다시 시작되고 있다고 표시됩니다. 이는 애플리케이션이 잘못 구성되었거나 외부 리소스에 액세스할 수 있는 권한이 없는 경우 흔히 발생하는 오류 시나리오입니다.

    이 경우 Flink 대시보드의 예외 탭에서 문제의 원인을 확인하세요.

실행 중인 애플리케이션의 메트릭을 살펴보세요.

MyApplication페이지의 Amazon CloudWatch 지표 섹션에서 실행 중인 애플리케이션의 몇 가지 기본 지표를 볼 수 있습니다.

지표를 보려면
  1. 새로 고침 버튼 옆의 드롭다운 목록에서 10초를 선택합니다.

  2. 애플리케이션이 실행 중이고 정상이면 가동 시간 지표가 계속 증가하는 것을 볼 수 있습니다.

  3. 전체 재시작 지표는 0이어야 합니다. 값이 증가하는 경우 구성에 문제가 있을 수 있습니다. 문제를 조사하려면 Flink 대시보드의 예외 탭을 검토하세요.

  4. 정상 애플리케이션에서는 실패한 체크포인트 수 지표가 0이어야 합니다.

    참고

    이 대시보드에는 5분 단위의 고정된 지표 집합이 표시됩니다. 대시보드의 모든 메트릭을 사용하여 사용자 지정 애플리케이션 대시보드를 만들 수 있습니다. CloudWatch

Kinesis 스트림의 출력 데이터 관찰

Python 스크립트 또는 Kinesis 데이터 생성기를 사용하여 여전히 데이터를 입력에 게시하고 있는지 확인하십시오.

이전에 이미 수행한 것과 마찬가지로 이제 의 데이터 뷰어를 사용하여 Apache Flink용 관리 서비스에서 실행되는 애플리케이션의 출력을 관찰할 수 있습니다. https://console.aws.amazon.com/kinesis/

출력을 보려면
  1. /kinesis에서 Kinesis 콘솔을 엽니다. https://console.aws.amazon.com

  2. 지역이 이 자습서를 실행하는 데 사용하는 지역과 동일한지 확인하십시오. 기본적으로 이 지역은 미국 동부-1미국 동부 (버지니아 북부) 입니다. 필요한 경우 지역을 변경하십시오.

  3. 데이터 스트림을 선택합니다.

  4. 관찰하려는 스트림을 선택합니다. 본 자습서에서는 ExampleOutputStream를 사용합니다.

  5. 데이터 뷰어 탭을 선택합니다.

  6. 샤드를 선택하고 최신 버전을 시작 위치로 유지한 다음 레코드 가져오기를 선택합니다. “이 요청에 대한 레코드를 찾을 수 없습니다.” 라는 오류가 표시될 수 있습니다. 그렇다면 레코드 가져오기 재시도를 선택하십시오. 스트림에 게시된 최신 레코드가 표시됩니다.

  7. 데이터 열에서 값을 선택하여 레코드 콘텐츠를 JSON 형식별로 검사합니다.

애플리케이션을 중지합니다.

애플리케이션을 중지하려면 이라는 이름의 Apache Flink용 관리 서비스 애플리케이션의 콘솔 페이지로 이동합니다. MyApplication

애플리케이션을 중지하려면
  1. 작업 드롭다운 목록에서 중지를 선택합니다.

  2. 애플리케이션 세부 정보의 상태가 에서 RunningStopping 전환된 다음 애플리케이션이 완전히 Ready 중지된 시점으로 전환됩니다.

    참고

    Python 스크립트 또는 Kinesis 데이터 생성기에서 입력 스트림으로 데이터를 보내는 것도 중단하는 것을 잊지 마십시오.

다음 단계

리소스 정리 AWS