AWS Athena
Since Camel 3.4
Only producer is supported
The AWS2 Athena component supports running queries with AWS Athena and working with results.
Prerequisites
You must have a valid Amazon Web Services developer account, and be signed up to use Amazon Athena. More information is available at AWS Athena.
URI Format
aws2-athena://label[?options]
You can append query options to the URI in the following format, ?options=value&option2=value&…
Configuring Options
Camel components are configured on two separate levels:
-
component level
-
endpoint level
Configuring Component Options
The component level is the highest level which holds general and common configurations that are inherited by the endpoints. For example a component may have security settings, credentials for authentication, urls for network connection and so forth.
Some components only have a few options, and others may have many. Because components typically have pre configured defaults that are commonly used, then you may often only need to configure a few options on a component; or none at all.
Configuring components can be done with the Component DSL, in a configuration file (application.properties|yaml), or directly with Java code.
Configuring Endpoint Options
Where you find yourself configuring the most is on endpoints, as endpoints often have many options, which allows you to configure what you need the endpoint to do. The options are also categorized into whether the endpoint is used as consumer (from) or as a producer (to), or used for both.
Configuring endpoints is most often done directly in the endpoint URI as path and query parameters. You can also use the Endpoint DSL as a type safe way of configuring endpoints.
A good practice when configuring options is to use Property Placeholders, which allows to not hardcode urls, port numbers, sensitive information, and other settings. In other words placeholders allows to externalize the configuration from your code, and gives more flexibility and reuse.
The following two sections lists all the options, firstly for the component followed by the endpoint.
Component Options
The AWS Athena component supports 30 options, which are listed below.
Name | Description | Default | Type |
---|---|---|---|
Amazon AWS Access Key. |
String |
||
Autowired The AmazonAthena instance to use as the client. |
AthenaClient |
||
The component configuration. |
Athena2Configuration |
||
The Athena database to use. |
String |
||
Milliseconds before the next poll for query execution status. See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. |
2000 |
long |
|
The encryption type to use when storing query results in S3. One of SSE_S3, SSE_KMS, or CSE_KMS. Enum values:
|
EncryptionOption |
||
Include useful trace information at the beginning of queries as an SQL comment (prefixed with --). |
false |
boolean |
|
Milliseconds before the first poll for query execution status. See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. |
1000 |
long |
|
For SSE-KMS and CSE-KMS, this is the KMS key ARN or ID. |
String |
||
Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel’s routing error handlers. Beware that when the first message is processed then creating and starting the producer may take a little time and prolong the total processing time of the processing. |
false |
boolean |
|
Maximum number of times to attempt a query. Set to 1 to disable retries. See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. |
1 |
int |
|
Max number of results to return for the given operation (if supported by the Athena API endpoint). If not set, will use the Athena API default for the given operation. |
Integer |
||
Pagination token to use in the case where the response from the previous request was truncated. |
String |
||
The Athena API function to call. Enum values:
|
startQueryExecution |
Athena2Operations |
|
The location in Amazon S3 where query results are stored, such as s3://path/to/query/bucket/. Ensure this value ends with a forward slash ('/'). |
String |
||
How query results should be returned. One of StreamList (default - return a GetQueryResultsIterable that can page through all results), SelectList (returns at most 1,000 rows at a time, plus a NextToken value as a header than can be used for manual pagination of results), S3Pointer (return an S3 path pointing to the results). Enum values:
|
StreamList |
Athena2OutputType |
|
To define a proxy host when instantiating the Athena client. |
String |
||
To define a proxy port when instantiating the Athena client. |
Integer |
||
To define a proxy protocol when instantiating the Athena client. Enum values:
|
HTTPS |
Protocol |
|
The unique ID identifying the query execution. |
String |
||
The SQL query to run. Except for simple queries, prefer setting this as the body of the Exchange or as a header using Athena2Constants.QUERY_STRING to avoid having to deal with URL encoding issues. |
String |
||
The region in which Athena client needs to work. When using this parameter, the configuration will expect the lowercase name of the region (for example ap-east-1). You’ll need to use the name Region.EU_WEST_1.id(). |
String |
||
Reset the waitTimeout countdown in the event of a query retry. If set to true, potential max time spent waiting for queries is equal to waitTimeout x maxAttempts. See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. |
true |
boolean |
|
Optional comma separated list of error types to retry the query for. Use 'retryable' to retry all retryable failure conditions (e.g. generic errors and resources exhausted), 'generic' to retry 'GENERIC_INTERNAL_ERROR' failures, 'exhausted' to retry queries that have exhausted resource limits, 'always' to always retry regardless of failure condition, or 'never' or null to never retry (default). See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. Enum values:
|
never |
String |
|
Amazon AWS Secret Key. |
String |
||
Set whether the Athena client should expect to load credentials through a default credentials provider or to expect static credentials to be passed in. |
false |
boolean |
|
Optional max wait time in millis to wait for a successful query completion. See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. |
0 |
long |
|
The workgroup to use for running the query. |
String |
||
Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching type, which then gets configured on the component. This can be used for automatic configuring JDBC data sources, JMS connection factories, AWS Clients, etc. |
true |
boolean |
|
A unique string to ensure issues queries are idempotent. It is unlikely you will need to set this. |
String |
Endpoint Options
The AWS Athena endpoint is configured using URI syntax:
aws2-athena:label
with the following path and query parameters:
Query Parameters (28 parameters)
Name | Description | Default | Type |
---|---|---|---|
Amazon AWS Access Key. |
String |
||
Autowired The AmazonAthena instance to use as the client. |
AthenaClient |
||
The Athena database to use. |
String |
||
Milliseconds before the next poll for query execution status. See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. |
2000 |
long |
|
The encryption type to use when storing query results in S3. One of SSE_S3, SSE_KMS, or CSE_KMS. Enum values:
|
EncryptionOption |
||
Include useful trace information at the beginning of queries as an SQL comment (prefixed with --). |
false |
boolean |
|
Milliseconds before the first poll for query execution status. See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. |
1000 |
long |
|
For SSE-KMS and CSE-KMS, this is the KMS key ARN or ID. |
String |
||
Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel’s routing error handlers. Beware that when the first message is processed then creating and starting the producer may take a little time and prolong the total processing time of the processing. |
false |
boolean |
|
Maximum number of times to attempt a query. Set to 1 to disable retries. See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. |
1 |
int |
|
Max number of results to return for the given operation (if supported by the Athena API endpoint). If not set, will use the Athena API default for the given operation. |
Integer |
||
Pagination token to use in the case where the response from the previous request was truncated. |
String |
||
The Athena API function to call. Enum values:
|
startQueryExecution |
Athena2Operations |
|
The location in Amazon S3 where query results are stored, such as s3://path/to/query/bucket/. Ensure this value ends with a forward slash ('/'). |
String |
||
How query results should be returned. One of StreamList (default - return a GetQueryResultsIterable that can page through all results), SelectList (returns at most 1,000 rows at a time, plus a NextToken value as a header than can be used for manual pagination of results), S3Pointer (return an S3 path pointing to the results). Enum values:
|
StreamList |
Athena2OutputType |
|
To define a proxy host when instantiating the Athena client. |
String |
||
To define a proxy port when instantiating the Athena client. |
Integer |
||
To define a proxy protocol when instantiating the Athena client. Enum values:
|
HTTPS |
Protocol |
|
The unique ID identifying the query execution. |
String |
||
The SQL query to run. Except for simple queries, prefer setting this as the body of the Exchange or as a header using Athena2Constants.QUERY_STRING to avoid having to deal with URL encoding issues. |
String |
||
The region in which Athena client needs to work. When using this parameter, the configuration will expect the lowercase name of the region (for example ap-east-1). You’ll need to use the name Region.EU_WEST_1.id(). |
String |
||
Reset the waitTimeout countdown in the event of a query retry. If set to true, potential max time spent waiting for queries is equal to waitTimeout x maxAttempts. See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. |
true |
boolean |
|
Optional comma separated list of error types to retry the query for. Use 'retryable' to retry all retryable failure conditions (e.g. generic errors and resources exhausted), 'generic' to retry 'GENERIC_INTERNAL_ERROR' failures, 'exhausted' to retry queries that have exhausted resource limits, 'always' to always retry regardless of failure condition, or 'never' or null to never retry (default). See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. Enum values:
|
never |
String |
|
Amazon AWS Secret Key. |
String |
||
Set whether the Athena client should expect to load credentials through a default credentials provider or to expect static credentials to be passed in. |
false |
boolean |
|
Optional max wait time in millis to wait for a successful query completion. See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. |
0 |
long |
|
The workgroup to use for running the query. |
String |
||
A unique string to ensure issues queries are idempotent. It is unlikely you will need to set this. |
String |
Required Athena component options
You have to provide the amazonAthenaClient in the Registry or your accessKey and secretKey to access the AWS Athena service.
Examples
Producer Examples
For example, to run a simple query, wait up to 60 seconds for completion, and log the results:
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");
Similarly, running the query and returning a path to the results in S3:
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");
Message headers evaluated by the Athena producer
Message headers listed here override the corresponding query parameters listed in "Query Parameters".
Operation | Header | Type | Description | Required |
---|---|---|---|---|
All |
|
|
The operation to perform. Permitted values are getQueryExecution, getQueryResults, listQueryExecutions, startQueryExecution. Default is startQueryExecution. |
No |
startQueryExecution |
|
|
The Athena database to use. |
No |
getQueryExecution, getQueryResults |
|
|
The unique ID identifying the query execution. |
No |
listQueryExecutions, startQueryExecution |
|
|
The workgroup to use for running the query. |
No |
getQueryResults, listQueryExecutions |
|
|
Pagination token to use in the case where the response from the previous request was truncated. |
No |
getQueryResults, listQueryExecutions |
|
|
Max number of results to return for the given operation (if supported by the Athena API endpoint). If not set, will use the Athena API default for the given operation. |
No |
startQueryExecution |
|
|
Include useful trace information at the beginning of queries as an SQL comment (prefixed with "--"). |
No |
startQueryExecution |
|
|
The location in Amazon S3 where query results are stored, such as s3://path/to/query/bucket/. Ensure this value ends with a forward slash ('/'). |
No |
getQueryResults |
|
|
How query results should be returned. One of StreamList (default - return a GetQueryResultsIterable that can page through all results), SelectList (returns at most 1,000 rows at a time, plus a NextToken value as a header than can be used for manual pagination of results), S3Pointer (return an S3 path pointing to the results). |
No |
startQueryExecution |
|
|
A unique string to ensure issues queries are idempotent. It is unlikely you will need to set this. |
No |
startQueryExecution |
|
|
The SQL query to run. Except for simple queries, prefer setting this as the |
No |
startQueryExecution |
|
|
The encryption type to use when storing query results in S3. One of SSE_S3, SSE_KMS, or CSE_KMS. |
No |
startQueryExecution |
|
|
For SSE-KMS and CSE-KMS, this is the KMS key ARN or ID. |
No |
startQueryExecution |
|
|
Optional max wait time in millis to wait for a successful query completion. See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. |
No |
startQueryExecution |
|
|
Milliseconds before the first poll for query execution status. See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. |
No |
startQueryExecution |
|
|
Milliseconds before the next poll for query execution status. See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. |
No |
startQueryExecution |
|
|
Maximum number of times to attempt a query. Set to 1 to disable retries. See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. |
No |
startQueryExecution |
|
|
Optional comma separated list of error types to retry the query for. Use 'retryable' to retry all retryable failure conditions (e.g. generic errors and resources exhausted), 'generic' to retry 'GENERIC_INTERNAL_ERROR' failures, 'exhausted' to retry queries that have exhausted resource limits, 'always' to always retry regardless of failure condition, or 'never' or null to never retry (default). See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. |
No |
startQueryExecution |
|
|
Reset the waitTimeout countdown in the event of a query retry. If set to true, potential max time spent waiting for queries is equal to waitTimeout x maxAttempts. See the section 'Waiting for Query Completion and Retrying Failed Queries' to learn more. |
No |
Message headers set by the Athena producer
Operation | Header | Type | Description |
---|---|---|---|
getQueryExecution, getQueryResults, startQueryExecution |
|
|
The unique ID identifying the query execution. |
getQueryExecution, getQueryResults, startQueryExecution |
|
|
The state of the query execution. One of |
getQueryExecution, getQueryResults, startQueryExecution |
|
|
The location in Amazon S3 where query results are stored, such as s3://path/to/query/bucket/. |
getQueryExecution, getQueryResults, startQueryExecution |
|
|
The unique ID identifying the query execution. |
getQueryResults, listQueryExecutions |
|
|
Pagination token to use in the case where the response from the previous request was truncated. |
startQueryExecution |
|
|
Total number of attempts made to run the query. Will be greater than 1 if the query is retried. |
startQueryExecution |
|
|
Total time in millis taken in startQueryExecution (mostly relevant when waiting for query completion within startQueryExecution). |
Static credentials vs Default Credential Provider
You have the possibility of avoiding the usage of explicit static credentials, by specifying the useDefaultCredentialsProvider option and set it to true.
-
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.
For more information about this you can look at AWS credentials documentation
Athena Producer operations
The Camel-AWS Athena component provides the following operation on the producer side:
-
getQueryExecution
-
getQueryResults
-
listQueryExecutions
-
startQueryExecution
Advanced AmazonAthena configuration
If your Camel Application is running behind a firewall or if you need to
have more control over the AthenaClient
instance configuration, you can
create your own instance and refer to it in your Camel aws2-athena component configuration:
from("aws2-athena://MyQuery?amazonAthenaClient=#client&...")
.to("mock:result");
Overriding query parameters with message headers
Message headers listed in "Message headers evaluated by the Athena producer" override the corresponding query parameters listed in "Query Parameters".
For example:
from("direct:start")
.setHeader(Athena2Constants.OUTPUT_LOCATION, constant("s3://other/location/"))
.to("aws2-athena:label?outputLocation=s3://foo/bar/")
.to("mock:result");
Will cause the output location to be s3://other/location/
.
Athena Producer Operation examples
-
getQueryExecution: this operation returns information about a query given its query execution ID
from("direct:start")
.to("aws2-athena://label?operation=getQueryExecution&queryExecutionId=11111111-1111-1111-1111-111111111111")
.to("mock:result");
The preceding example will yield an Athena QueryExecution in the body.
The getQueryExecution operation also supports retreiving the query execution ID from a header
(CamelAwsAthenaQueryExecutionId
), and since startQueryExecution sets the same header upon starting a query,
these operations can be used together:
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");
The preceding example will yield an Athena QueryExecution in the body for the query that was just started.
-
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
- the default - returns a
GetQueryResultsIterable
in the body that can page through all results:
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");
The output of StreamList can be processed in various ways:
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(new Processor() {
@Override
public void process(Exchange 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");
The preceding example will print the results of the query as CSV to the console.
SelectList
- returns a
GetQueryResponse
in the body containing at most 1,000 rows, plus the NextToken value as a header (CamelAwsAthenaNextToken
),
which can be used for manual pagination of results:
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");
The preceding example will return a
GetQueryResponse
in the body plus the NextToken value as a header (CamelAwsAthenaNextToken
), which can be used to manually page
through the results 1,000 rows at a time.
S3Pointer
- return an S3 path (e.g. s3://bucket/path/
) pointing to the results:
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");
The preceding example will return an
S3 path (e.g. s3://bucket/path/
) in the body pointing to the results. The path will also be set in a header
(CamelAwsAthenaOutputLocation
).
-
listQueryExecutions: this operation returns a list of query execution IDs
from("direct:start")
.to("aws2-athena://label?operation=listQueryExecutions")
.to("mock:result");
The preceding example will return a list of query executions in the body, plus the NextToken value as a
header (CamelAwsAthenaNextToken
) than can be used for manual pagination of results.
-
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:
from("direct:start")
.setBody(constant("SELECT 1"))
.to("aws2-athena://label?operation=startQueryExecution&outputLocation=s3://bucket/path/")
.to("mock:result");
The preceding example will start the query SELECT 1
and configure the
results to be saved to s3://bucket/path/
, but will not wait for the query
to complete.
from("direct:start")
.setBody(constant("SELECT 1"))
.to("aws2-athena://label?operation=startQueryExecution&waitTimeout=60000&outputLocation=s3://bucket/path/")
.to("mock:result");
The preceding example will start a query and wait up to 60 seconds for it to reach a status that indicates it is complete (one of SUCCEEDED, FAILED, CANCELLED, or UNKNOWN_TO_SDK_VERSION). Upon failure, the query would not be retried.
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");
The preceding example will start a query and wait up to 60 seconds for it to reach
a status that indicates it is complete (one of SUCCEEDED, FAILED, CANCELLED,
or UNKNOWN_TO_SDK_VERSION). Upon failure, the query would be automatically
retried up to 2 more times if the failure state indicates the query may
succeed upon retry (Athena queries that fail with states such as GENERIC_INTERNAL_ERROR
or "resource limit exhaustion" will sometimes succeed if retried). While waiting
for the query to complete, the query status would first be checked after an
initial delay of 10 seconds, and subsequently every 1 second until the query completes.
Putting it all together
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");
The preceding example will start the query and wait up to 60 seconds for it to complete. Upon completion, getQueryResults put the results of the query into the body of the message for further processing.
For the sake of completeness, a similar outcome could be achieved with the following:
from("direct:start")
.setBody(constant("SELECT 1"))
.to("aws2-athena://label?operation=startQueryExecution&outputLocation=s3://bucket/path/")
.loopDoWhile(simple("${header." + Athena2Constants.QUERY_EXECUTION_STATE + "} != 'SUCCEEDED'"))
.delay(1_000)
.to("aws2-athena://label?operation=getQueryExecution")
.end()
.to("aws2-athena://label?operation=getQueryResults&outputType=StreamList")
.to("mock:result");
Caution: the preceding example would block indefinitely, however, if the query did not complete with a status of SUCCEEDED.
Dependencies
Maven users will need to add the following dependency to their pom.xml.
pom.xml
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-aws2-athena</artifactId>
<version>${camel-version}</version>
</dependency>
where ${camel-version}
must be replaced by the actual version of Camel.
Spring Boot Auto-Configuration
When using aws2-athena with Spring Boot make sure to use the following Maven dependency to have support for auto configuration:
<dependency>
<groupId>org.apache.camel.springboot</groupId>
<artifactId>camel-aws2-athena-starter</artifactId>
<version>x.x.x</version>
<!-- use the same version as your Camel core version -->
</dependency>
The component supports 31 options, which are listed below.