AWS Athena

원문 보기 위키 갱신 2026.10.11

AWS Athena (camel-aws2-athena)

AWS2 Athena 컴포넌트는 AWS Athena에서 쿼리를 실행하고 그 결과를 다룰 수 있게 해 주는 컴포넌트예요. aws2-athena://label URI로 엔드포인트를 만들고, operation으로 startQueryExecution·getQueryExecution·getQueryResults·listQueryExecutions 연산을 고를 수 있어요. 쿼리 완료 대기(waitTimeout), 실패 재시도(retry, maxAttempts), S3 출력 위치 같은 옵션들이 제공돼요.

출처: 문서

본문

Camel 3.4부터 지원

프로듀서만 지원돼요

AWS2 Athena 컴포넌트는 AWS Athena로 쿼리를 실행하고 그 결과를 다룰 수 있게 해 줘요.

사전 준비

유효한 Amazon Web Services 개발자 계정이 있고 Amazon Athena를 사용하도록 가입되어 있어야 해요. 자세한 내용은 AWS Athena에서 확인하세요.

URI Format

aws2-athena://label[?options]

URI 뒤에 다음 형식으로 쿼리 옵션을 붙일 수 있어요: ?options=value&option2=value&…

Configuring Options

Camel 컴포넌트는 서로 다른 두 단계에서 설정돼요:

component level
endpoint level

Configuring Component Options

컴포넌트 단계에서는 엔드포인트들이 물려받는 일반적이고 공통된 설정을 해요. 가장 상위의 설정 단계죠.

예컨대 컴포넌트에는 보안 설정, 인증에 쓰는 credentials, 네트워크 연결을 위한 URL 같은 것들이 있을 수 있어요.

컴포넌트마다 옵션이 몇 개뿐인 경우도 있고 많은 경우도 있어요. 컴포넌트는 보통 자주 쓰는 기본값이 미리 준비되어 있어서, 몇 개의 옵션만 설정하거나 아예 하나도 건드리지 않아도 되는 경우가 많아요.

컴포넌트를 구성하는 방법은 이렇게 있어요:

the Component DSL.
in a configuration file (application.properties, *.yaml files, etc).
directly in the Java code.

Configuring Endpoint Options

엔드포인트는 옵션이 많은 편이라 설정에 시간을 더 쓰게 돼요. 이 옵션들은 엔드포인트가 하고 싶은 동작을 맞춤 설정하는 데 도움을 줘요. 옵션은 엔드포인트가 소비자(from)로 쓰이는지, 프로듀서(to)로 쓰이는지, 둘 다인지에 따라 분류되기도 해요.

엔드포인트 설정은 대부분 엔드포인트 URI의 _path_와 query 파라미터로 직접 이뤄져요. Java에서 엔드포인트와 데이터 포맷을 _타입 세이프_하게 설정하고 싶다면 Endpoint DSL과 DataFormat DSL을 쓸 수도 있어요.

옵션을 설정할 때 좋은 습관 하나는 Property Placeholders를 활용하는 거예요.

프로퍼티 플레이스홀더는 이런 장점들이 있어요:

They help prevent using hardcoded urls, port numbers, sensitive information, and other settings.
They allow externalizing the configuration from the code.
They help the code to become more flexible and reusable.

지금부터 두 절에 걸쳐 모든 옵션을 나열해요. 먼저 컴포넌트 옵션이고, 그다음이 엔드포인트 옵션이에요.

Component Options

AWS Athena 컴포넌트는 아래 나열한 옵션들을 지원해요.

Name	Description	Default	Type
configuration (producer)	컴포넌트 설정이에요.		Athena2Configuration
database (producer)	사용할 Athena 데이터베이스예요.		String
delay (producer)	다음 폴링(쿼리 실행 상태 조회) 전에 기다리는 밀리초예요. 자세한 내용은 '쿼리 완료 대기 및 실패한 쿼리 재시도' 섹션을 참고하세요.	2000	long
initialDelay (producer)	첫 폴링(쿼리 실행 상태 조회) 전에 기다리는 밀리초예요. 자세한 내용은 '쿼리 완료 대기 및 실패한 쿼리 재시도' 섹션을 참고하세요.	1000	long
lazyStartProducer (producer)	프로듀서를 lazy(첫 메시지가 올 때)로 시작할지 여부예요. lazy로 시작하면 프로듀서가 시작 시점에 실패해서 라우트 시작까지 실패하는 상황에서도 CamelContext와 라우트를 띄울 수 있어요. 시작을 lazy로 미루면 시작 실패를 라우팅 중에 Camel의 라우팅 에러 핸들러로 처리할 수 있죠. 다만 첫 메시지를 처리할 때는 프로듀서를 만들고 시작하는 데 시간이 조금 걸려서 전체 처리 시간이 길어질 수 있다는 점은 감안하세요.	false	boolean
maxAttempts (producer)	쿼리를 시도할 최대 횟수예요. 1로 설정하면 재시도하지 않아요. 자세한 내용은 '쿼리 완료 대기 및 실패한 쿼리 재시도' 섹션을 참고하세요.	1	int
maxResults (producer)	해당 연산에 대해 반환할 최대 결과 개수예요(Athena API 엔드포인트가 지원하는 경우). 설정하지 않으면 해당 연산의 Athena API 기본값을 사용해요.		Integer
nextToken (producer)	이전 요청의 응답이 잘렸을 때(truncated) 사용할 페이지네이션 토큰이에요.		String
operation (producer)
호출할 Athena API 함수예요.

Enum 값:

getQueryExecution
getQueryResults
listQueryExecutions
startQueryExecution

쿼리 결과를 어떻게 반환할지 정해요. StreamList(기본값 — 전체 결과를 페이징하며 넘겨줄 수 있는 GetQueryResultsIterable을 반환), SelectList(한 번에 최대 1000행을 반환하고, 결과를 수동으로 페이징할 때 쓸 NextToken 값을 헤더로 함께 돌려줌), S3Pointer(결과를 가리키는 S3 경로를 반환) 중 하나예요.

Enum 값:

StreamList
SelectList
S3Pointer

Athena 클라이언트가 동작할 리전이에요. 이 파라미터를 쓰면 설정은 리전의 소문자 이름(예: ap-east-1)을 기대해요.

Enum 값:

ap-south-2
ap-south-1
eu-south-1
eu-south-2
us-gov-east-1
me-central-1
il-central-1
ca-central-1
eu-central-1
us-iso-west-1
eu-central-2
eu-isoe-west-1
us-west-1
us-west-2
af-south-1
eu-north-1
eu-west-3
eu-west-2
eu-west-1
ap-northeast-3
ap-northeast-2
ap-northeast-1
me-south-1
sa-east-1
ap-east-1
cn-north-1
ca-west-1
us-gov-west-1
ap-southeast-1
ap-southeast-2
us-iso-east-1
ap-southeast-3
ap-southeast-4
us-east-1
us-east-2
cn-northwest-1
us-isob-east-1
aws-global
aws-cn-global
aws-us-gov-global
aws-iso-global
aws-iso-b-global

쿼리를 재시도할 오류 타입의 쉼표 구분 목록(선택)이에요. 'retryable'은 재시도 가능한 모든 실패 조건(일반 오류, 자원 고갈 등)을 재시도하고, 'generic'은 'GENERIC_INTERNAL_ERROR' 실패를, 'exhausted'는 자원 한도를 소진한 쿼리를, 'always'는 실패 조건과 상관없이 항상, 'never'나 null은 절대 재시도하지 않는다(기본값)는 뜻이에요. 자세한 내용은 '쿼리 완료 대기 및 실패한 쿼리 재시도' 섹션을 참고하세요.

Enum 값:

never
always
retryable
exhausted
generic

Athena 클라이언트를 인스턴스화할 때 프록시 프로토콜을 지정해요.

Enum 값:

HTTP
HTTPS

쿼리 결과를 S3에 저장할 때 쓸 암호화 방식이에요. SSE_S3, SSE_KMS, CSE_KMS 중 하나예요.

Enum 값:

SSE_S3
SSE_KMS
CSE_KMS
null

Endpoint Options

AWS Athena 엔드포인트는 다음 URI 문법으로 설정해요:

aws2-athena:label

아래 path 파라미터와 query 파라미터를 사용해요:

Path Parameters

Name	Description	Default	Type
label (producer)	필수 논리적 이름이에요.		String

Query Parameters

Name	Description	Default	Type
database (producer)	사용할 Athena 데이터베이스예요.		String
delay (producer)	다음 폴링(쿼리 실행 상태 조회) 전에 기다리는 밀리초예요. 자세한 내용은 '쿼리 완료 대기 및 실패한 쿼리 재시도' 섹션을 참고하세요.	2000	long
initialDelay (producer)	첫 폴링(쿼리 실행 상태 조회) 전에 기다리는 밀리초예요. 자세한 내용은 '쿼리 완료 대기 및 실패한 쿼리 재시도' 섹션을 참고하세요.	1000	long
maxAttempts (producer)	쿼리를 시도할 최대 횟수예요. 1로 설정하면 재시도하지 않아요. 자세한 내용은 '쿼리 완료 대기 및 실패한 쿼리 재시도' 섹션을 참고하세요.	1	int
maxResults (producer)	해당 연산에 대해 반환할 최대 결과 개수예요(Athena API 엔드포인트가 지원하는 경우). 설정하지 않으면 해당 연산의 Athena API 기본값을 사용해요.		Integer
nextToken (producer)	이전 요청의 응답이 잘렸을 때(truncated) 사용할 페이지네이션 토큰이에요.		String
operation (producer)
호출할 Athena API 함수예요.

Enum 값:

getQueryExecution
getQueryResults
listQueryExecutions
startQueryExecution

Enum 값:

StreamList
SelectList
S3Pointer

Athena 클라이언트가 동작할 리전이에요. 이 파라미터를 쓰면 설정은 리전의 소문자 이름(예: ap-east-1)을 기대해요.

Enum 값:

ap-south-2
ap-south-1
eu-south-1
eu-south-2
us-gov-east-1
me-central-1
il-central-1
ca-central-1
eu-central-1
us-iso-west-1
eu-central-2
eu-isoe-west-1
us-west-1
us-west-2
af-south-1
eu-north-1
eu-west-3
eu-west-2
eu-west-1
ap-northeast-3
ap-northeast-2
ap-northeast-1
me-south-1
sa-east-1
ap-east-1
cn-north-1
ca-west-1
us-gov-west-1
ap-southeast-1
ap-southeast-2
us-iso-east-1
ap-southeast-3
ap-southeast-4
us-east-1
us-east-2
cn-northwest-1
us-isob-east-1
aws-global
aws-cn-global
aws-us-gov-global
aws-iso-global
aws-iso-b-global

Enum 값:

never
always
retryable
exhausted
generic

Athena 클라이언트를 인스턴스화할 때 프록시 프로토콜을 지정해요.

Enum 값:

HTTP
HTTPS

쿼리 결과를 S3에 저장할 때 쓸 암호화 방식이에요. SSE_S3, SSE_KMS, CSE_KMS 중 하나예요.

Enum 값:

SSE_S3
SSE_KMS
CSE_KMS
null

Message Headers

AWS Athena 컴포넌트는 아래 나열한 메시지 헤더들을 지원해요:

Name	Description	Default	Type
CamelAwsAthenaOperation (all) Constant: `OPERATION`
수행할 연산이에요. 허용되는 값은 getQueryExecution, getQueryResults, listQueryExecutions, startQueryExecution이에요.

Enum 값:

getQueryExecution
getQueryResults
listQueryExecutions
startQueryExecution

| startQueryExecution | Athena2Operations | | CamelAwsAthenaDatabase (startQueryExecution) Constant: DATABASE | 사용할 Athena 데이터베이스예요. | | String | | CamelAwsAthenaQueryExecutionId (getQueryExecution getQueryResults startQueryExecution) Constant: QUERY_EXECUTION_ID | 쿼리 실행을 식별하는 유일한 ID예요. | | String | | CamelAwsAthenaWorkGroup (listQueryExecutions startQueryExecution) Constant: WORK_GROUP | 쿼리를 실행할 때 사용할 워크그룹이에요. | | String | | CamelAwsAthenaNextToken (getQueryResults listQueryExecutions) Constant: NEXT_TOKEN | 이전 요청의 응답이 잘렸을 때(truncated) 사용할 페이지네이션 토큰이에요. | | String | | CamelAwsAthenaMaxResults (getQueryResults listQueryExecutions) Constant: MAX_RESULTS | 해당 연산에 대해 반환할 최대 결과 개수예요(Athena API 엔드포인트가 지원하는 경우). 설정하지 않으면 해당 연산의 Athena API 기본값을 사용해요. | | Integer | | CamelAwsAthenaIsTruncated (getQueryResults listQueryExecutions) Constant: IS_TRUNCATED | 응답에 결과가 더 있는지(즉 잘렸는지) 여부예요. true면 NEXT_TOKEN 헤더로 다음 페이지를 가져오세요. | | Boolean | | CamelAwsAthenaIncludeTrace (startQueryExecution) Constant: INCLUDE_TRACE | 쿼리 시작 부분에 유용한 트레이스 정보를 SQL 주석(-- 접두사)으로 포함해요. | | boolean | | CamelAwsAthenaOutputLocation (getQueryExecution getQueryResults startQueryExecution) Constant: OUTPUT_LOCATION | 쿼리 결과가 저장되는 Amazon S3 위치예요. s3://path/to/query/bucket/ 같은 식이에요. 이 값은 슬래시('/')로 끝나야 해요. | | String | | CamelAwsAthenaOutputType (getQueryResults) Constant: OUTPUT_TYPE |

쿼리 결과를 어떻게 반환할지 정해요. StreamList(기본값 — 전체 결과를 페이징하며 넘겨줄 수 있는 GetQueryResultsIterable을 반환), SelectList(한 번에 최대 1,000행을 반환하고, 결과를 수동으로 페이징할 때 쓸 NextToken 값을 헤더로 함께 돌려줌), S3Pointer(결과를 가리키는 S3 경로를 반환) 중 하나예요.

Enum 값:

StreamList
SelectList
S3Pointer

| | Athena2OutputType | | CamelAwsAthenaQueryExecutionState (getQueryExecution getQueryResults startQueryExecution) Constant: QUERY_EXECUTION_STATE |

쿼리 실행의 상태예요.

Enum 값:

QUEUED
RUNNING
SUCCEEDED
FAILED
CANCELLED
null

| | QueryExecutionState | | CamelAwsAthenaClientRequestToken (startQueryExecution) Constant: CLIENT_REQUEST_TOKEN | 쿼리 요청이 멱등(idempotent)하게 처리되도록 보장하는 유일한 문자열이에요. 설정할 일은 거의 없을 거예요. | | String | | CamelAwsAthenaQueryString (startQueryExecution) Constant: QUERY_STRING | 실행할 SQL 쿼리예요. 단순한 쿼리가 아니라면 URL 인코딩 문제를 피하려고 이 옵션 대신 Exchange 본문이나 이 헤더로 설정하는 게 좋아요. | | String | | CamelAwsAthenaEncryptionOption (startQueryExecution) Constant: ENCRYPTION_OPTION |

쿼리 결과를 S3에 저장할 때 쓸 암호화 방식이에요.

Enum 값:

SSE_S3
SSE_KMS
CSE_KMS
null

| | EncryptionOption | | CamelAwsAthenaKmsKey (startQueryExecution) Constant: KMS_KEY | SSE-KMS와 CSE-KMS에서 쓰는 KMS 키 ARN 또는 ID예요. | | String | | CamelAwsAthenaWaitTimeout (startQueryExecution) Constant: WAIT_TIMEOUT | 쿼리가 성공적으로 끝날 때까지 기다릴 최대 대기 시간(밀리초, 선택)이에요. 자세한 내용은 '쿼리 완료 대기 및 실패한 쿼리 재시도' 섹션을 참고하세요. | | long | | CamelAwsAthenaInitialDelay (startQueryExecution) Constant: INITIAL_DELAY | 첫 폴링(쿼리 실행 상태 조회) 전에 기다리는 밀리초예요. 자세한 내용은 '쿼리 완료 대기 및 실패한 쿼리 재시도' 섹션을 참고하세요. | | long | | CamelAwsAthenaDelay (startQueryExecution) Constant: DELAY | 다음 폴링(쿼리 실행 상태 조회) 전에 기다리는 밀리초예요. 자세한 내용은 '쿼리 완료 대기 및 실패한 쿼리 재시도' 섹션을 참고하세요. | | long | | CamelAwsAthenaMaxAttempts (startQueryExecution) Constant: MAX_ATTEMPTS | 쿼리를 시도할 최대 횟수예요. 1로 설정하면 재시도하지 않아요. 자세한 내용은 '쿼리 완료 대기 및 실패한 쿼리 재시도' 섹션을 참고하세요. | | int | | CamelAwsAthenaRetry (startQueryExecution) Constant: RETRY | 쿼리를 재시도할 오류 타입의 쉼표 구분 목록(선택)이에요. 'retryable'을 쓰면 재시도 가능한 모든 실패 조건(일반 오류, 자원 고갈 등)을 재시도하고, 'generic'은 'GENERIC_INTERNAL_ERROR' 실패를, 'exhausted'는 자원 한도를 소진한 쿼리를, 'always'는 실패 조건과 상관없이 항상, 'never'나 null은 절대 재시도하지 않는다(기본값)는 뜻이에요. 자세한 내용은 '쿼리 완료 대기 및 실패한 쿼리 재시도' 섹션을 참고하세요. | | String | | CamelAwsAthenaResetWaitTimeoutOnRetry (startQueryExecution) Constant: RESET_WAIT_TIMEOUT_ON_RETRY | 쿼리를 재시도할 때 waitTimeout 카운트다운을 초기화할지 여부예요. true로 하면 쿼리 대기에 쓸 수 있는 최대 시간은 waitTimeout x maxAttempts가 돼요. 자세한 내용은 '쿼리 완료 대기 및 실패한 쿼리 재시도' 섹션을 참고하세요. | | boolean | | CamelAwsAthenaStartQueryExecutionAttempts (startQueryExecution) Constant: START_QUERY_EXECUTION_ATTEMPTS | 쿼리를 실행하기 위해 시도한 총 횟수예요. 재시도가 있으면 1보다 커요. | | int | | CamelAwsAthenaStartQueryExecutionElapsedMillis (startQueryExecution) Constant: START_QUERY_EXECUTION_ELAPSED_MILLIS | startQueryExecution에 걸린 총 시간(밀리초)이에요. startQueryExecution 안에서 쿼리 완료를 기다릴 때 주로 의미가 있어요. | | long |

필수 Athena 컴포넌트 옵션

AWS Athena 서비스에 접근하려면 amazonAthenaClient를 레지스트리에 제공하거나 accessKey와 secretKey를 제공해야 해요.

Examples

Producer Examples

예컨대 단순한 쿼리를 실행하고, 완료를 최대 60초까지 기다리고, 결과를 로그로 남기려면 이렇게 해요:

Java
XML
YAML

from("direct:start")
    .setBody(constant("SELECT 1"))
    .to("aws2-athena://label?waitTimeout=60000&outputLocation=s3://bucket/path/")
    .to("aws2-athena://label?operation=getQueryResults&outputType=StreamList")
    .split(body()).streaming()
    .to("log:out")
    .to("mock:result");

<route>
  <from uri="direct:start"/>
  <setBody>
    <constant>SELECT 1</constant>
  </setBody>
  <to uri="aws2-athena://label?waitTimeout=60000&amp;outputLocation=s3://bucket/path/"/>
  <to uri="aws2-athena://label?operation=getQueryResults&amp;outputType=StreamList"/>
  <split streaming="true">
    <simple>${body}</simple>
    <to uri="log:out"/>
    <to uri="mock:result"/>
  </split>
</route>

- route:
    from:
      uri: direct:start
      steps:
        - setBody:
            constant: SELECT 1
        - to:
            uri: aws2-athena://label
            parameters:
              waitTimeout: "60000"
              outputLocation: s3://bucket/path/
        - to:
            uri: aws2-athena://label
            parameters:
              operation: getQueryResults
              outputType: StreamList
        - split:
            simple: "${body}"
            streaming: true
            steps:
              - to:
                  uri: log:out
              - to:
                  uri: mock:result

비슷하게, 쿼리를 실행하고 그 결과가 있는 S3 경로를 돌려받으려면 이렇게 해요:

Java
XML
YAML

from("direct:start")
    .setBody(constant("SELECT 1"))
    .to("aws2-athena://label?waitTimeout=60000&outputLocation=s3://bucket/path/")
    .to("aws2-athena://label?operation=getQueryResults&outputType=S3Pointer")
    .to("mock:result");

<route>
  <from uri="direct:start"/>
  <setBody>
    <constant>SELECT 1</constant>
  </setBody>
  <to uri="aws2-athena://label?waitTimeout=60000&amp;outputLocation=s3://bucket/path/"/>
  <to uri="aws2-athena://label?operation=getQueryResults&amp;outputType=S3Pointer"/>
  <to uri="mock:result"/>
</route>

- route:
    from:
      uri: direct:start
      steps:
        - setBody:
            constant: SELECT 1
        - to:
            uri: aws2-athena://label
            parameters:
              waitTimeout: "60000"
              outputLocation: s3://bucket/path/
        - to:
            uri: aws2-athena://label
            parameters:
              operation: getQueryResults
              outputType: S3Pointer
        - to:
            uri: mock:result

Static credentials, Default Credential Provider and Profile Credentials Provider

useDefaultCredentialsProvider 옵션을 지정하고 true로 설정하면 명시적인 정적(static) credentials를 쓰지 않도록 할 수 있어요.

Default Credentials Provider의 평가 순서는 다음과 같아요:

Java system properties - aws.accessKeyId and aws.secretKey.
Environment variables - AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY.
Web Identity Token from AWS STS.
The shared credentials and config files.
Amazon ECS container credentials - loaded from the Amazon ECS if the environment variable AWS_CONTAINER_CREDENTIALS_RELATIVE_URI is set.
Amazon EC2 Instance profile credentials.

useProfileCredentialsProvider 옵션을 true로, profileCredentialsName을 프로파일 이름으로 지정하면 Profile Credentials Provider를 사용할 수도 있어요.

정적(static), 기본(default), 프로파일(profile) credentials는 동시에 딱 하나만 쓸 수 있어요.

자세한 내용은 AWS credentials 문서를 참고하세요.

Athena Producer operations

Camel-AWS Athena 컴포넌트는 프로듀서 쪽에서 다음 연산을 제공해요:

getQueryExecution
getQueryResults
listQueryExecutions
startQueryExecution

Advanced AmazonAthena configuration

Camel 애플리케이션이 방화벽 뒤에서 돌거나 AthenaClient 인스턴스 설정을 더 세밀하게 통제해야 한다면, 직접 인스턴스를 만들고 Camel aws2-athena 컴포넌트 설정에서 그 인스턴스를 참조하게 할 수 있어요:

Java
XML
YAML

from("direct:start")
    .to("aws2-athena://MyQuery?amazonAthenaClient=#client")
    .to("mock:result");

<route>
  <from uri="direct:start"/>
  <to uri="aws2-athena://MyQuery?amazonAthenaClient=#client"/>
  <to uri="mock:result"/>
</route>

- route:
    from:
      uri: direct:start
      steps:
        - to:
            uri: aws2-athena://MyQuery
            parameters:
              amazonAthenaClient: "#client"
        - to:
            uri: mock:result

Overriding query parameters with message headers

'Athena 프로듀서가 평가하는 메시지 헤더'에 나열된 메시지 헤더는 '쿼리 파라미터'에 나열된 대응하는 쿼리 파라미터를 덮어써요.

예를 들면 이렇게 돼요:

Java
XML
YAML

from("direct:start")
     .setHeader(Athena2Constants.OUTPUT_LOCATION, constant("s3://other/location/"))
     .to("aws2-athena:label?outputLocation=s3://foo/bar/")
     .to("mock:result");

<route>
  <from uri="direct:start"/>
  <setHeader name="CamelAwsAthenaOutputLocation">
    <constant>s3://other/location/</constant>
  </setHeader>
  <to uri="aws2-athena:label?outputLocation=s3://foo/bar/"/>
  <to uri="mock:result"/>
</route>

- route:
    from:
      uri: direct:start
      steps:
        - setHeader:
            name: CamelAwsAthenaOutputLocation
            constant: "s3://other/location/"
        - to:
            uri: aws2-athena:label
            parameters:
              outputLocation: "s3://foo/bar/"
        - to:
            uri: mock:result

이렇게 하면 출력 위치가 s3://other/location/로 정해져요.

Athena Producer Operation examples

getQueryExecution: this operation returns information about a query given its query execution ID
Java
XML
YAML

from("direct:start")
    .to("aws2-athena://label?operation=getQueryExecution&queryExecutionId=11111111-1111-1111-1111-111111111111")
    .to("mock:result");

<route>
    <from uri="direct:start"/>
    <to uri="aws2-athena://label?operation=getQueryExecution&amp;queryExecutionId=11111111-1111-1111-1111-111111111111"/>
    <to uri="mock:result"/>
</route>

- route:
    from:
      uri: direct:start
      steps:
        - to:
            uri: aws2-athena://label
            parameters:
              operation: getQueryExecution
              queryExecutionId: 11111111-1111-1111-1111-111111111111
        - to:
            uri: mock:result

앞선 예제는 본문에 Athena QueryExecution을 담아 돌려줘요.

getQueryExecution 연산은 쿼리 실행 ID를 헤더(CamelAwsAthenaQueryExecutionId)에서 가져올 수도 있어요. startQueryExecution이 같은 헤더를 설정해 주기 때문에, 쿼리를 시작한 뒤에는 이 두 연산을 이렇게 함께 쓸 수 있어요:

Java
XML
YAML

from("direct:start")
    .setBody(constant("SELECT 1"))
    .to("aws2-athena://label?operation=startQueryExecution&outputLocation=s3://bucket/path/")
    .to("aws2-athena://label?operation=getQueryExecution")
    .to("mock:result");

<route>
  <from uri="direct:start"/>
  <setBody>
    <constant>SELECT 1</constant>
  </setBody>
  <to uri="aws2-athena://label?operation=startQueryExecution&amp;outputLocation=s3://bucket/path/"/>
  <to uri="aws2-athena://label?operation=getQueryExecution"/>
  <to uri="mock:result"/>
</route>

- route:
    from:
      uri: direct:start
      steps:
        - setBody:
            constant: SELECT 1
        - to:
            uri: aws2-athena://label
            parameters:
              operation: startQueryExecution
              outputLocation: s3://bucket/path/
        - to:
            uri: aws2-athena://label
            parameters:
              operation: getQueryExecution
        - to:
            uri: mock:result

앞선 예제는 방금 시작한 쿼리의 Athena QueryExecution을 본문에 담아 돌려줘요.

getQueryResults: this operation returns the results of a query that has succeeded. The results are returned in the body in one of three formats.

StreamList — 기본값 — 은 전체 결과를 페이징하며 넘겨줄 수 있는 GetQueryResultsIterable을 본문에 담아 돌려줘요:

Java
XML
YAML

from("direct:start")
    .setBody(constant("SELECT 1"))
    .to("aws2-athena://label?operation=startQueryExecution&waitTimeout=60000&outputLocation=s3://bucket/path/")
    .to("aws2-athena://label?operation=getQueryResults&outputType=StreamList")
    .to("mock:result");

<route>
  <from uri="direct:start"/>
  <setBody>
    <constant>SELECT 1</constant>
  </setBody>
  <to uri="aws2-athena://label?operation=startQueryExecution&amp;waitTimeout=60000&amp;outputLocation=s3://bucket/path/"/>
  <to uri="aws2-athena://label?operation=getQueryResults&amp;outputType=StreamList"/>
  <to uri="mock:result"/>
</route>

- route:
    from:
      uri: direct:start
      steps:
        - setBody:
            constant: SELECT 1
        - to:
            uri: aws2-athena://label
            parameters:
              operation: startQueryExecution
              waitTimeout: "60000"
              outputLocation: s3://bucket/path/
        - to:
            uri: aws2-athena://label
            parameters:
              operation: getQueryResults
              outputType: StreamList
        - to:
            uri: mock:result

StreamList의 출력은 여러 가지 방법으로 처리할 수 있어요:

Java 전용: AWS SDK GetQueryResultsResponse 처리

from("direct:start")
    .setBody(constant("SELECT * FROM (VALUES (1, 'a'), (2, 'b')) AS t (id, name)"))
    .to("aws2-athena://label?operation=startQueryExecution&waitTimeout=60000&outputLocation=s3://bucket/path/")
    .to("aws2-athena://label?operation=getQueryResults&outputType=StreamList")
    .split(body()).streaming()
    .process(exchange -> {
        GetQueryResultsResponse page = exchange.getMessage().getBody(GetQueryResultsResponse.class);
        for (Row row : page.resultSet().rows()) {
          String line = row.data().stream()
              .map(Datum::varCharValue)
              .collect(Collectors.joining(","));
          System.out.println(line);
        }
    })
    .to("mock:result");

앞선 예제는 쿼리 결과를 CSV로 콘솔에 출력해요.

SelectList — 본문에 최대 1,000행을 담은 GetQueryResponse를 돌려주고, NextToken 값을 헤더(CamelAwsAthenaNextToken)로 함께 줘서 결과를 수동으로 페이징할 수 있게 해 줘요:

Java
XML
YAML

from("direct:start")
    .setBody(constant("SELECT 1"))
    .to("aws2-athena://label?operation=startQueryExecution&waitTimeout=60000&outputLocation=s3://bucket/path/")
    .to("aws2-athena://label?operation=getQueryResults&outputType=SelectList")
    .to("mock:result");

<route>
  <from uri="direct:start"/>
  <setBody>
    <constant>SELECT 1</constant>
  </setBody>
  <to uri="aws2-athena://label?operation=startQueryExecution&amp;waitTimeout=60000&amp;outputLocation=s3://bucket/path/"/>
  <to uri="aws2-athena://label?operation=getQueryResults&amp;outputType=SelectList"/>
  <to uri="mock:result"/>
</route>

- route:
    from:
      uri: direct:start
      steps:
        - setBody:
            constant: SELECT 1
        - to:
            uri: aws2-athena://label
            parameters:
              operation: startQueryExecution
              waitTimeout: "60000"
              outputLocation: s3://bucket/path/
        - to:
            uri: aws2-athena://label
            parameters:
              operation: getQueryResults
              outputType: SelectList
        - to:
            uri: mock:result

앞선 예제는 본문에 GetQueryResponse를 담아 돌려주고, NextToken 값도 헤더(CamelAwsAthenaNextToken)로 함께 줘서 결과를 1,000행씩 수동으로 넘겨 볼 수 있게 해 줘요.

S3Pointer — 결과를 가리키는 S3 경로(예: s3://bucket/path/)를 돌려줘요:

Java
XML
YAML

from("direct:start")
    .setBody(constant("SELECT 1"))
    .to("aws2-athena://label?operation=startQueryExecution&waitTimeout=60000&outputLocation=s3://bucket/path/")
    .to("aws2-athena://label?operation=getQueryResults&outputType=S3Pointer")
    .to("mock:result");

<route>
  <from uri="direct:start"/>
  <setBody>
    <constant>SELECT 1</constant>
  </setBody>
  <to uri="aws2-athena://label?operation=startQueryExecution&amp;waitTimeout=60000&amp;outputLocation=s3://bucket/path/"/>
  <to uri="aws2-athena://label?operation=getQueryResults&amp;outputType=S3Pointer"/>
  <to uri="mock:result"/>
</route>

- route:
    from:
      uri: direct:start
      steps:
        - setBody:
            constant: SELECT 1
        - to:
            uri: aws2-athena://label
            parameters:
              operation: startQueryExecution
              waitTimeout: "60000"
              outputLocation: s3://bucket/path/
        - to:
            uri: aws2-athena://label
            parameters:
              operation: getQueryResults
              outputType: S3Pointer
        - to:
            uri: mock:result

앞선 예제는 본문에 결과를 가리키는 S3 경로(예: s3://bucket/path/)를 담아 돌려줘요. 이 경로는 헤더(CamelAwsAthenaOutputLocation)에도 설정돼요.

listQueryExecutions: this operation returns a list of query execution IDs
Java
XML
YAML

from("direct:start")
    .to("aws2-athena://label?operation=listQueryExecutions")
    .to("mock:result");

<route>
    <from uri="direct:start"/>
    <to uri="aws2-athena://label?operation=listQueryExecutions"/>
    <to uri="mock:result"/>
</route>

- route:
    from:
      uri: direct:start
      steps:
        - to:
            uri: aws2-athena://label
            parameters:
              operation: listQueryExecutions
        - to:
            uri: mock:result

앞선 예제는 본문에 쿼리 실행 목록을 돌려주고, 결과를 수동으로 페이징할 때 쓸 NextToken 값도 헤더(CamelAwsAthenaNextToken)로 함께 줘요.

startQueryExecution: this operation starts the execution of a query. It supports waiting for the query to complete before proceeding, and retrying the query based on a set of configurable failure conditions:
Java
XML
YAML

from("direct:start")
    .setBody(constant("SELECT 1"))
    .to("aws2-athena://label?operation=startQueryExecution&outputLocation=s3://bucket/path/")
    .to("mock:result");

<route>
    <from uri="direct:start"/>
    <setBody>
        <constant>SELECT 1</constant>
    </setBody>
    <to uri="aws2-athena://label?operation=startQueryExecution&amp;outputLocation=s3://bucket/path/"/>
    <to uri="mock:result"/>
</route>

- route:
    from:
      uri: direct:start
      steps:
        - setBody:
            constant: SELECT 1
        - to:
            uri: aws2-athena://label
            parameters:
              operation: startQueryExecution
              outputLocation: "s3://bucket/path/"
        - to:
            uri: mock:result

앞선 예제는 SELECT 1 쿼리를 시작하고 결과가 s3://bucket/path/에 저장되도록 설정하지만, 쿼리가 끝나기를 기다리지는 않아요.

Java
XML
YAML

from("direct:start")
    .setBody(constant("SELECT 1"))
    .to("aws2-athena://label?operation=startQueryExecution&waitTimeout=60000&outputLocation=s3://bucket/path/")
    .to("mock:result");

<route>
  <from uri="direct:start"/>
  <setBody>
    <constant>SELECT 1</constant>
  </setBody>
  <to uri="aws2-athena://label?operation=startQueryExecution&amp;waitTimeout=60000&amp;outputLocation=s3://bucket/path/"/>
  <to uri="mock:result"/>
</route>

- route:
    from:
      uri: direct:start
      steps:
        - setBody:
            constant: SELECT 1
        - to:
            uri: aws2-athena://label
            parameters:
              operation: startQueryExecution
              waitTimeout: "60000"
              outputLocation: s3://bucket/path/
        - to:
            uri: mock:result

앞선 예제는 쿼리를 시작하고 완료를 뜻하는 상태(SUCCEEDED, FAILED, CANCELLED, UNKNOWN_TO_SDK_VERSION 중 하나)에 도달할 때까지 최대 60초를 기다려요. 실패하면 쿼리를 재시도하지 않아요.

Java
XML
YAML

from("direct:start")
    .setBody(constant("SELECT 1"))
    .to("aws2-athena://label?operation=startQueryExecution&waitTimeout=60000&initialDelay=10000&delay=1000&maxAttempts=3&retry=retryable&outputLocation=s3://bucket/path/")
    .to("mock:result");

<route>
  <from uri="direct:start"/>
  <setBody>
    <constant>SELECT 1</constant>
  </setBody>
  <to uri="aws2-athena://label?operation=startQueryExecution&amp;waitTimeout=60000&amp;initialDelay=10000&amp;delay=1000&amp;maxAttempts=3&amp;retry=retryable&amp;outputLocation=s3://bucket/path/"/>
  <to uri="mock:result"/>
</route>

- route:
    from:
      uri: direct:start
      steps:
        - setBody:
            constant: SELECT 1
        - to:
            uri: aws2-athena://label
            parameters:
              operation: startQueryExecution
              waitTimeout: "60000"
              initialDelay: "10000"
              delay: "1000"
              maxAttempts: 3
              retry: retryable
              outputLocation: s3://bucket/path/
        - to:
            uri: mock:result

앞선 예제는 쿼리를 시작하고 완료를 뜻하는 상태(SUCCEEDED, FAILED, CANCELLED, UNKNOWN_TO_SDK_VERSION 중 하나)에 도달할 때까지 최대 60초를 기다려요. 실패했을 때, 그 실패 상태가 재시도하면 성공할 수 있음을 뜻한다면 쿼리가 자동으로 최대 두 번 더 재시도돼요(GENERIC_INTERNAL_ERROR 같은 상태나 '자원 한도 고갈'로 실패한 Athena 쿼리는 재시도하면 가끔 성공하기도 해요). 쿼리 완료를 기다리는 동안에는 처음에 10초의 초기 지연 후 쿼리 상태를 확인하고, 그 뒤로는 쿼리가 끝날 때까지 1초마다 확인해요.

Putting it all together

Java
XML
YAML

from("direct:start")
    .setBody(constant("SELECT 1"))
    .to("aws2-athena://label?waitTimeout=60000&maxAttempts=3&retry=retryable&outputLocation=s3://bucket/path/")
    .to("aws2-athena://label?operation=getQueryResults&outputType=StreamList")
    .to("mock:result");

<route>
  <from uri="direct:start"/>
  <setBody>
    <constant>SELECT 1</constant>
  </setBody>
  <to uri="aws2-athena://label?waitTimeout=60000&amp;maxAttempts=3&amp;retry=retryable&amp;outputLocation=s3://bucket/path/"/>
  <to uri="aws2-athena://label?operation=getQueryResults&amp;outputType=StreamList"/>
  <to uri="mock:result"/>
</route>

- route:
    from:
      uri: direct:start
      steps:
        - setBody:
            constant: SELECT 1
        - to:
            uri: aws2-athena://label
            parameters:
              waitTimeout: "60000"
              maxAttempts: 3
              retry: retryable
              outputLocation: s3://bucket/path/
        - to:
            uri: aws2-athena://label
            parameters:
              operation: getQueryResults
              outputType: StreamList
        - to:
            uri: mock:result

앞선 예제는 쿼리를 시작하고 완료를 최대 60초까지 기다려요. 완료되면 getQueryResults가 쿼리 결과를 메시지 본문에 넣어서 후속 처리로 넘겨줘요.

완전성을 위해, 비슷한 결과를 다음과 같이 얻을 수도 있어요:

Java
XML
YAML

from("direct:start")
    .setBody(constant("SELECT 1"))
    .to("aws2-athena://label?operation=startQueryExecution&outputLocation=s3://bucket/path/")
    .loopDoWhile(simple("${header.CamelAwsAthenaQueryExecutionState} != 'SUCCEEDED'"))
      .delay(1_000)
      .to("aws2-athena://label?operation=getQueryExecution")
    .end()
    .to("aws2-athena://label?operation=getQueryResults&outputType=StreamList")
    .to("mock:result");

<route>
  <from uri="direct:start"/>
  <setBody>
    <constant>SELECT 1</constant>
  </setBody>
  <to uri="aws2-athena://label?operation=startQueryExecution&amp;outputLocation=s3://bucket/path/"/>
  <loop doWhile="true">
    <simple>${header.CamelAwsAthenaQueryExecutionState} != 'SUCCEEDED'</simple>
    <delay>
      <constant>1000</constant>
    </delay>
    <to uri="aws2-athena://label?operation=getQueryExecution"/>
  </loop>
  <to uri="aws2-athena://label?operation=getQueryResults&amp;outputType=StreamList"/>
  <to uri="mock:result"/>
</route>

- route:
    from:
      uri: direct:start
      steps:
        - setBody:
            constant: SELECT 1
        - to:
            uri: aws2-athena://label
            parameters:
              operation: startQueryExecution
              outputLocation: s3://bucket/path/
        - loop:
            doWhile: true
            simple: "${header.CamelAwsAthenaQueryExecutionState} != 'SUCCEEDED'"
            steps:
              - delay:
                  constant: 1000
              - to:
                  uri: aws2-athena://label
                  parameters:
                    operation: getQueryExecution
        - to:
            uri: aws2-athena://label
            parameters:
              operation: getQueryResults
              outputType: StreamList
        - to:
            uri: mock:result

주의: 앞선 예제는 쿼리가 SUCCEEDED 상태로 끝나지 않으면 무한히 멈춰 있게 돼요.

Dependencies

Maven 사용자라면 pom.xml에 다음 의존성을 추가해야 해요.

pom.xml

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-aws2-athena</artifactId>
    <version>${camel-version}</version>
</dependency>

${camel-version}은 실제 Camel 버전으로 바꿔야 해요.

더 알아보기 (Learn more)

컴포넌트·엔드포인트 옵션은 위의 Component Options / Endpoint Options 표를 참고하고, 쿼리 대기·재시도 동작은 Examples의 startQueryExecution 예제들이 잘 보여줘요.
Athena 클라이언트를 직접 구성해야 한다면 'Advanced AmazonAthena configuration' 섹션처럼 amazonAthenaClient=#client로 커스텀 클라이언트를 붙일 수 있어요.
인증은 accessKey/secretKey 정적 credentials, useDefaultCredentialsProvider, useProfileCredentialsProvider 세 가지 중 하나만 동시에 쓸 수 있다는 점을 기억하세요.

AWS Athena

AWS Athena (camel-aws2-athena)

본문

URI Format

Configuring Options

Configuring Component Options

Configuring Endpoint Options

Component Options

Endpoint Options

Path Parameters

Query Parameters

Message Headers

Examples

Producer Examples

Static credentials, Default Credential Provider and Profile Credentials Provider

Athena Producer operations

Advanced AmazonAthena configuration

Overriding query parameters with message headers

Athena Producer Operation examples

Putting it all together

Dependencies

더 알아보기 (Learn more)

함께 읽는 실무 이야기