### Install Dependencies Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/python/data-generator/README.md Install the required dependencies for the data generator application. ```bash python -m pip install -r requirements.txt ``` -------------------------------- ### Install PyFlink and Dependencies Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/python/PythonDependencies/README.md Installs the Apache Flink Python library and other project requirements using pip within a virtual environment. ```bash pip install apache-flink==1.20.0 ``` ```bash pip install -r requirements.txt ``` -------------------------------- ### Install PyFlink and Dependencies Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/python/PackagedPythonDependencies/README.md Install the PyFlink library and other Python dependencies using pip. It's recommended to use a virtual environment. The `--target` option installs dependencies into a specific directory, and `--platform` ensures compatibility with the target execution environment. ```bash pip install apache-flink==1.20.0 pip install -r requirements.txt pip install -r requirements.txt --target=dep/ --platform=manylinux2014_x86_64 --only-binary=:all: ``` -------------------------------- ### Kafka Console Producer Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/java/KafkaConfigProviders/Kafka-mTLS-Keystore-ConfigProviders/docs/step-by-step.md Start the Kafka console producer to send messages to a specified topic. Ensure you are in the Kafka installation directory. ```bash cd /home/ec2-user/kafka bin/kafka-console-producer.sh --broker-list $brokers --topic ``` -------------------------------- ### Sample Data Format Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/python/data-generator/README.md An example of the data format generated by the stock data generator. ```json {'event_time': '2024-05-28T19:53:17.497201', 'ticker': 'AMZN', 'price': 42.88} ``` -------------------------------- ### Clone Flink Examples Repository Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/java/KafkaConfigProviders/Kafka-mTLS-Keystore-ConfigProviders/docs/step-by-step.md Clone the repository containing the Flink application code. This is a prerequisite for building the custom keystore configuration provider. ```bash git clone https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples cd Java/KafkaCustomKeystoreWithConfigProviders ``` -------------------------------- ### Example Stock Price Data Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/java/KinesisSourceDeaggregation/kpl-producer/README.md This is an example of the JSON data format for stock price records that the producer can generate and publish. ```json { 'event_time': '2024-05-28T19:53:17.497201', 'ticker': 'AMZN', 'price': 42.88 } ``` -------------------------------- ### Dockerfile for PyFlink 1.15 Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/python/LocalDevelopmentOnAppleSilicon/README.md This Dockerfile installs Python 3.8, boto3, and AWS CLI, based on Flink 1.15 documentation. It's designed to build an image for the 'linux/amd64' platform to avoid dependency issues on Apple Silicon. ```dockerfile FROM flink:1.15.4 ARG PYTHON_VERSION=3.8.9 RUN apt-get update -y && \ apt-get install -y build-essential libssl-dev zlib1g-dev libbz2-dev libffi-dev && \ wget https://www.python.org/ftp/python/${PYTHON_VERSION}/Python-${PYTHON_VERSION}.tgz && \ tar -xvf Python-${PYTHON_VERSION}.tgz && \ cd Python-${PYTHON_VERSION} && \ ./configure --without-tests --enable-shared && \ make -j6 && \ make install && \ ldconfig /usr/local/lib && \ cd .. && rm -f Python-${PYTHON_VERSION}.tgz && rm -rf Python-${PYTHON_VERSION} && \ ln -s /usr/local/bin/python3 /usr/local/bin/python && \ apt-get clean && \ rm -rf /var/lib/apt/lists/* RUN pip3 install apache-flink==1.15.4 boto3 awscli ``` -------------------------------- ### Create Kafka Topic Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/java/KafkaConfigProviders/Kafka-mTLS-Keystore-ConfigProviders/docs/step-by-step.md Use the kafka-topics utility to create a new topic on your MSK cluster. Specify the bootstrap server, topic name, number of partitions, and replication factor. ```bash /home/ec2-user/kafka/bin/kafka-topics.sh --bootstrap-server $brokers --create --topic --partitions --replication-factor ``` -------------------------------- ### Configure Java Kafka Producer Properties Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/java/KafkaConfigProviders/Kafka-mTLS-Keystore-ConfigProviders/docs/step-by-step.md Create a `producer.properties_msk` file with bootstrap server details, keystore password, and key password for mTLS authentication. ```bash cd /tmp/kafka rm producer.properties_msk cat < producer.properties_msk BOOTSTRAP_SERVERS_CONFIG=$brokers SSL_KEYSTORE_PASSWORD_CONFIG= SSL_KEY_PASSWORD_CONFIG= EOF ``` -------------------------------- ### Sample S3 Output Listing Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/java/S3Sink/README.md Lists the files written to the S3 bucket. Ensure to replace `` with your actual bucket name. ```shell 2023-10-13 13:29:33 74 output/2023-10-13--13/_part-84f63a7b-ba44-46f9-96a7-8fdf248767c8-0_tmp_d252d9b4-6382-4d6f-9e65-3fa3f058e9c5 2023-10-13 13:29:33 109 output/2023-10-13--13/_part-a87f5ad1-b920-463d-a2bc-bfc3c8ee2c81-0_tmp_fd16d35a-1297-4eb0-98cd-25301b1d12ba ``` -------------------------------- ### List S3 Table Buckets Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/java/Iceberg/S3TableSink/README.md If you have an existing S3 table bucket, use this command to list them and retrieve the ARN for configuration. ```bash aws s3tables list-table-buckets ``` -------------------------------- ### Find PyFlink Home Directory Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/python/S3Sink/README.md Use this command to locate the PyFlink installation directory, which is necessary for installing local Flink dependencies. Ensure your virtual environment is activated. ```python import pyflink;import os;print(os.path.dirname(os.path.abspath(pyflink.__file__))) ``` -------------------------------- ### Compile and Run KPL Producer Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/java/KinesisSourceDeaggregation/kpl-producer/README.md Compile the project using Maven and run the generated JAR file. Specify the Kinesis stream name and region as command-line arguments. The 'sleep' parameter controls the delay between publishing records. ```bash mvn package ``` ```bash java -jar target/kpl-producer-1.0.jar --streamName --streamRegion --sleep 10 ``` -------------------------------- ### Sample Output Format Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/java/Windowing/README.md Illustrates the expected format of aggregated data output from the Flink job. ```text "AMZN",12.80,TimeWindow{start=123456789000, end=123456790000} ``` -------------------------------- ### Get Kafka Client Keystore DN Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/java/KafkaConfigProviders/Kafka-mTLS-Keystore-ConfigProviders/docs/step-by-step.md Use this command to retrieve the Distinguished Name (DN) from your Kafka client keystore, which is required for setting ACLs. ```bash keytool --list -v -keystore /tmp/kafka.client.keystore.jks|grep ip- ``` -------------------------------- ### Flink Iceberg Sink DataStream API Example Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/java/Iceberg/S3TableSink/README.md This Java code demonstrates setting up the Flink Iceberg Sink with S3 tables using the DataStream API. Ensure necessary dependencies are included in your pom.xml. ```java public class IcebergSinkExample { public static void main(String[] args) throws Exception { // Set up the streaming execution environment StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(10000); // Checkpoint every 10 seconds // Configure Iceberg catalog and table Configuration conf = new Configuration(); conf.setString("table.bucket.arn", "arn:aws:s3tables:us-east-1:111122223333:bucket/flink-example"); conf.setString("catalog.db", "default"); conf.setString("catalog.table", "prices_s3table"); conf.setString("partition.fields", "symbol"); conf.setString("operation", "append"); // Create a DataStream of Avro GenericRecords (simulating source data) DataStream dataStream = env.addSource(new DataGeneratorSource(100.0)); // Create the Iceberg Sink IcebergSink icebergSink = IcebergSink.forGenericRecords( new org.apache.iceberg.flink.CatalogLoader.S3Table("default", "prices_s3table", conf), dataStream.getType().getOrThrow(), conf ); // Write the data stream to the Iceberg sink dataStream.sinkTo(icebergSink); // Execute the Flink job env.execute("Flink Iceberg Sink Example"); } // Dummy DataGeneratorSource for demonstration private static class DataGeneratorSource implements SourceFunction { private final double recordsPerSecond; private volatile boolean isRunning = true; public DataGeneratorSource(double recordsPerSecond) { this.recordsPerSecond = recordsPerSecond; } @Override public void run(SourceContext ctx) throws Exception { long sleepTime = (long) (1000.0 / recordsPerSecond); while (isRunning) { // Simulate generating an Avro GenericRecord GenericRecord record = new org.apache.avro.generic.GenericData.Record(new org.apache.avro.Schema.Parser().parse("{\"type\": \"record\", \"name\": \"StockPrice\", \"fields\": [{\"name\": \"symbol\", \"type\": \"string\"}, {\"name\": \"price\", \"type\": \"double\"}]")); record.put("symbol", "IBM"); record.put("price", Math.random() * 100); ctx.collect(record); Thread.sleep(sleepTime); } } @Override public void cancel() { isRunning = false; } } } ``` -------------------------------- ### Set Python Requirements for Flink Runtime Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/python/PythonDependencies/README.md This code snippet demonstrates how to specify a requirements.txt file for the Flink runtime environment. Flink will automatically install any listed dependencies that are not already available. Ensure the requirements.txt file does not include PyFlink dependencies. ```python python_source_dir = str(pathlib.Path(__file__).parent) table_env.set_python_requirements(requirements_file_path="file:///" + python_source_dir + "/requirements.txt") ``` -------------------------------- ### Sample Output File Content Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/java/S3Sink/README.md Displays the JSON content of a typical output file written to the S3 bucket. ```json {"eventTime":1715863231917,"ticker":"MSFT","price":85.92103591154482} {"eventTime":1715863232397,"ticker":"AMZN","price":83.69439555402906} {"eventTime":1715863232398,"ticker":"AMZN","price":85.15101593687162} ``` -------------------------------- ### Retrieve Task Status of Flink Job Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/infrastructure/scripts/README.md Use this script to get the status of each task in your Flink job. It's helpful for automating operations and checking deployment success or runtime failures. The job is considered healthy and processing data when all tasks are 'RUNNING'. ```shell > ./task_status.sh MyApplication RUNNING RUNNING RUNNING ``` ```shell > ./task_status.sh MyApplication FAILED CANCELED ``` ```shell > ./task_status.sh MyApplication UNKNOWN ``` -------------------------------- ### Kinesis Data Stream Usage Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/python/data-generator/README.md Run the stock data generator to publish data to a Kinesis stream. If no stream name is specified, `ExampleInputStream` will be used. ```bash python stock.py ``` -------------------------------- ### Build Application JAR Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/python/PythonDependencies/README.md Compiles the Java/Scala part of the application and packages dependencies, including the Kinesis connector, into a JAR file. This step is required before local execution or deployment. ```bash mvn package ``` -------------------------------- ### Create S3 Table Bucket Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/java/Iceberg/S3TableSink/README.md Use this AWS CLI command to create a new S3 table bucket. The ARN returned is required for runtime configuration. ```bash aws s3tables create-table-bucket --name flink-example { "arn": "arn:aws:s3tables:us-east-1:111122223333:bucket/flink-example" } ``` -------------------------------- ### Generate mTLS Keystore and Certificates Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/java/KafkaConfigProviders/Kafka-mTLS-Keystore-ConfigProviders/docs/step-by-step.md Use the `AuthMSK` JAR to generate a private key, create a Java Keystore, store the private key, convert it to PEM, generate a CSR, obtain an issued certificate from ACM PCA, and store the certificate in the Keystore. Replace `` and `` with your specific values. ```java cd /tmp/kafka java -jar AuthMSK-1.0-SNAPSHOT.jar -caa -ksp -ksa msk -pem ``` -------------------------------- ### Configure Kafka Connector with mTLS Config Providers Source: https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/java/KafkaConfigProviders/Kafka-mTLS-Keystore-ConfigProviders/README.md Define the names of config providers and their implementation classes. Then, set properties for Keystore type, location (fetched from S3), and passwords (fetched from AWS Secrets Manager). ```java KafkaSourceBuilder builder = ... ... // define names of config providers: builder.setProperty("config.providers", "secretsmanager,s3import"); // provide implementation classes for each provider: builder.setProperty("config.providers.secretsmanager.class", "com.amazonaws.kafka.config.providers.SecretsManagerConfigProvider"); builder.setProperty("config.providers.s3import.class", "com.amazonaws.kafka.config.providers.S3ImportConfigProvider"); String region = appProperties.get("S3BucketRegion"); String keystoreS3Bucket = appProperties.get("KeystoreS3Bucket"); String keystoreS3Path = appProperties.get("KeystoreS3Path"); String keystorePassSecret = appProperties.get("KeystorePassSecret"); String keystorePassSecretField = appProperties.get("KeystorePassSecretField"); // region, etc.. builder.setProperty("config.providers.s3import.param.region", region); // properties builder.setProperty("ssl.keystore.type", "PKCS12"); builder.setProperty("ssl.keystore.location", "${s3import:" + region + ":" + keystoreS3Bucket + "/" + keystoreS3Path + "}"); builder.setProperty("ssl.keystore.password", "${secretsmanager:" + keystorePassSecret + ":" + keystorePassSecretField + "}"); builder.setProperty("ssl.key.password", "${secretsmanager:" + keystorePassSecret + ":" + keystorePassSecretField + "}"); ```