> For the complete documentation index, see [llms.txt](https://developer.harness.io/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://developer.harness.io/feature-management-experimentation/management-and-administration/integrations/guides/impression-integration-serverless.md).
# Set Up Impression Integration to AWS S3 Using Kinesis Firehose and Lambda Cloud Services
### Overview
This page describes how to setup all the AWS Cloud Serverless components to store Split Impressions data in AWS S3 in JSON format.
Split Admin site provide a webhook integration option to send the Impression data as they are injected into Split cloud, we will use the webhook to extract Impression data.
We will also use AWS API Gateway as the Webhook service, the API Gateway will forward the post request to AWS Lambda, which will extract the Impression information and send it to AWS Kinesis Firehose, which has a built-in integration with S3.
### Setup
1. To create the Kinesis Firehose service, login to AWS and select **Services** > **Kinesis** > **Data Firehose**.
2. Put the service name, make sure the `Direct PUT or other sources` option is set, and click **Next**.
3. Click **Next** again on the **Select a destination** page, make sure **S3** is selected, specify the S3 bucket for the target storage, and click **Next**.
4. Select the IAM Role needed to access the service, consult with an AWS expert for more info, in this example, we used a role that has a `AmazonKinesisFullAccess` policy, click **Next**, then click **Create delivery stream**.
5. The Second service needed is AWS Lambda, select **AWS** > **Lambda**, click on the **Create function** button, choose the **Author from scratch** option, and use `Java 8` for the runtime engine.
6. Next step is to create the Java code for Lambda service, we will use an **Eclipse Maven** project with the following dependencies:
```java
com.amazonaws
aws-lambda-java-core
1.0.0
com.amazonaws
aws-java-sdk-core
1.11.671
com.amazonaws
amazon-kinesis-client
1.11.2
org.json
json
20180813
```
7. Insert the code below in the class (the class name here is `WebhookConsumer.java`), make sure to replace the `awsAccessKey`, `awsSecretKey`, `kinesisStreamName`, and `awsRegion`. The code below will convert the JSON structure passed from the POST request to Dictionary objects, then loop through each Impression block and send a bulk post to the Kinesis Stream service.
```java
package splitsupport;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.json.JSONObject;
import com.amazonaws.ClientConfiguration;
import com.amazonaws.ClientConfigurationFactory;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicSessionCredentials;
import com.amazonaws.services.kinesis.AmazonKinesis;
import com.amazonaws.services.kinesis.AmazonKinesisClientBuilder;
import com.amazonaws.services.kinesis.model.PutRecordsRequest;
import com.amazonaws.services.kinesis.model.PutRecordsRequestEntry;
import com.amazonaws.services.kinesis.model.PutRecordsResult;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
public class WebhookConsumer implements RequestHandler {
@Override
public String handleRequest(Object input, Context context) {
String treatment="";
try {
String awsAccessKey="xxxxxxxxxxx";
String awsSecretKey="xxxxxxxxxxxxxxxxxxxxx";
String awsRegion="us-east-2";
String kinesisStreamName="SplitStream";
BasicSessionCredentials awsCredentials = new BasicSessionCredentials(awsAccessKey, awsSecretKey, sessionToken);
ClientConfiguration awsClientConfig = new ClientConfigurationFactory().getConfig();
AmazonKinesisClientBuilder clientBuilder = AmazonKinesisClientBuilder.standard();
clientBuilder.setRegion(awsRegion);
clientBuilder.setCredentials(new AWSStaticCredentialsProvider(awsCredentials));
clientBuilder.setClientConfiguration(awsClientConfig);
AmazonKinesis kinesisClient = clientBuilder.build();
List> impressions = StringToHash(input.toString());
PutRecordsRequest putRecordsRequest = new PutRecordsRequest();
putRecordsRequest.setStreamName(kinesisStreamName);
List putRecordsRequestEntryList = new ArrayList<>();
PutRecordsRequestEntry putRecordsRequestEntry = new PutRecordsRequestEntry();
for (Map imp1 : impressions) {
JSONObject messageJson = new JSONObject();
messageJson.put("key", imp1.get("key"));
messageJson.put("split", imp1.get("split"));
messageJson.put("environmentName", imp1.get("environmentName"));
messageJson.put("time", imp1.get("time"));
messageJson.put("label", imp1.get("label"));
messageJson.put("treatment", imp1.get("treatment"));
putRecordsRequestEntry.setData(ByteBuffer.wrap(messageJson.toString().getBytes()));
putRecordsRequestEntry.setPartitionKey("partition-2");
putRecordsRequestEntryList.add(putRecordsRequestEntry);
}
putRecordsRequest.setRecords(putRecordsRequestEntryList);
PutRecordsResult putRecordsResult = kinesisClient.putRecords(putRecordsRequest);
System.out.println("Put Result" + putRecordsResult);
treatment=putRecordsResult.toString();
} catch (Exception e) {
System.out.print("Exception: "+e.getMessage());
return "Exception: "+e.getMessage();
}
return treatment;
}
public static List> StringToHash(String decomp) {
List> arrayRecord = new ArrayList>();
if (decomp.contains("}, {")) {
for(final String entry : decomp.split("},")) {
Map oneRecord = new HashMap();
for(final String parts : entry.split(",")) {
final String[] record = parts.split("=");
String recordOne = record[0].replace("[{", "");
recordOne = recordOne.replace("{", "");
String recordTwo = record[1].replace("}]", "");
oneRecord.put(recordOne.trim(), recordTwo.trim());
}
arrayRecord.add(oneRecord);
}
} else {
Map oneRecord = new HashMap();
for(final String parts : decomp.split(",")) {
final String[] record = parts.split("=");
String recordOne = record[0].replace("[{", "");
String recordTwo = record[1].replace("}]", "");
oneRecord.put(recordOne.trim(), recordTwo.trim());
}
arrayRecord.add(oneRecord);
}
return arrayRecord;
}
}
```
8. Export a self-contained JAR file for your project, in Eclipse, right-click on the project name and select **Export**.
9. Select the **Runnable JAR File** option and click **Next**, then specify the JAR file name and click **Finish**.
10. Back on the AWS Lambda page, under the **Function code** section, update the **Handler** box with the following: `splitsupport.WebhookConsumer::handleRequest`. Then, upload the JAR file you just created:
11. The Third service needed is the API Gateway, select **Services** > **API Gateway**.
12. Click on the **Create API** button, use `Rest API` and `New API` options, type a name for your API service, and click **Create API**.
13. Click on your new created API Service, click the `Actions` list to add a POST Method Execution, select **Lambda Function** for the Integration type, and type the Lambda function name created previously.
14. Since Split zip the post body when calling the Webhook, we need to enable binary type for our API Gateway, under **Settings**, add `*/*` in **Binary Media Types**, and click **Save Changes**.
15. The last step is to create a Stage for your API Gateway, and generate the webhook URL under the **Stages** page, click **Create**.
16. Your Webhook service is ready now. From the FME navigation menu, click **FME Settings** and navigate to the **Integrations** page.
17. Locate the Outgoing Webhook (Impressions) integration and click **Add**.
18. Select which environment's impressions to export, paste the API Gateway URL, and click **Save**.
