Python용 Apache Flink용 매니지드 서비스 생성 및 실행 애플리케이션

데이터 스트림 (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 

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

   git clone https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples.git

2. ./python/GettingStarted 디렉터리로 이동합니다.

명령줄에서 응용 프로그램을 실행하려면

1. Conda와 같은 독립형 Python 환경 또는 Python Flink 라이브러리를 설치한 VirtualEnv 환경이 현재 활성화되어 있는지 확인하십시오.

2. mvn package적어도 한 번은 실행했는지 확인하세요.

3. IS_LOCAL = true 환경 변수를 설정합니다.

   $ export IS_LOCAL=true

4. 응용 프로그램을 일반 Python 스크립트로 실행합니다.

   $python main.py

내에서 응용 프로그램을 실행하려면 IDE

1. 다음 구성으로 main.py 스크립트를 IDE 실행하도록 구성하십시오.

   1. Conda와 같은 독립형 Python 환경 또는 라이브러리를 설치한 VirtualEnv PyFlink 환경을 사용하십시오.

   2. AWS 자격 증명을 사용하여 입력 및 출력 Kinesis 데이터 스트림에 액세스할 수 있습니다.

   3. IS_LOCAL = true을 설정합니다.

2. 실행 구성을 설정하는 정확한 프로세스는 사용자에 따라 IDE 다르며 다양합니다.

3. 를 IDE 설정했으면 Python 스크립트를 실행하고 응용 프로그램이 실행되는 IDE 동안 사용자가 제공하는 도구를 사용하십시오.

애플리케이션 코드 JAR 파일을 업로드하려면

1. 에서 Amazon S3 콘솔을 엽니다 https://console.aws.amazon.com/s3/.

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

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

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

5. 이전 단계에서 생성된.zip 파일로 이동합니다. target/managed-flink-pyflink-getting-started-1.0.0.zip

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

1. /flink에서 아파치 플링크용 매니지드 서비스 콘솔을 여십시오. https://console.aws.amazon.com

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

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

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

   • 스트림 처리 응용 프로그램을 설정하는 방법 선택에서 처음부터 만들기를 선택합니다.

   • 아파치 플링크 구성, 애플리케이션 플링크 버전의 경우 아파치 플링크 1.19를 선택하십시오.

   • 애플리케이션 구성의 경우:

     • 애플리케이션 명칭에 MyApplication을 입력합니다.

     • 설명에 My Python test app를 입력합니다.

     • 애플리케이션 리소스 액세스에서 필수 정책을 사용하여 kinesis-analytics- MyApplication -us-east-1 IAM 역할 생성/업데이트를 선택합니다.

   • 애플리케이션용 템플릿 설정의 경우:

     • 템플릿의 경우 개발을 선택합니다.

   • 스트리밍 애플리케이션 생성을 선택합니다.

IAM정책을 편집하여 S3 버킷 권한을 추가하려면

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 객체 경로에는 managed-flink-pyflink-getting-started-1.0.0.zip를 입력합니다.

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

4. 런타임 속성으로 이동하여 다른 모든 설정의 기본값을 유지합니다.

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

   그룹 ID	키	값
   InputStream0	stream.name	ExampleInputStream
   InputStream0	flink.stream.initpos	LATEST
   InputStream0	aws.region	us-east-1
   OutputStream0	stream.name	ExampleOutputStream
   OutputStream0	aws.region	us-east-1
   kinesis.analytics.flink.run.options	python	main.py
   kinesis.analytics.flink.run.options	jarfile	lib/pyflink-dependencies.jar

6. 다른 섹션은 수정하지 말고 변경 내용 저장을 선택합니다.

애플리케이션을 실행하려면

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

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

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

대시보드 열기

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

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

   참고

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

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

지표를 보려면

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

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

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

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

   참고

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

출력을 보려면

1. /kinesis에서 Kinesis 콘솔을 엽니다. https://console.aws.amazon.com

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

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

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

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

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

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

애플리케이션을 중지하려면

1. 작업 드롭다운 목록에서 중지를 선택합니다.

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

   참고

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