Add mysql source

Author: Hoang Phan

python/sources/mysql_cdc/.gitignore

@@ -0,0 +1,10 @@
+**/ref/
+**/obj/
+**/bin/
+DotSettings.user
+**DotSettings.user
+ca.cert
+.idea/
+__pycache__/
+certificates/
+state/

python/sources/mysql_cdc/README.md

@@ -0,0 +1,39 @@
+# MySQL CDC
+
+This connector demonstrates how to capture changes to a MySQL database table (using CDC) and publish the change events to a Kafka topic using MySQL binary log replication.
+
+## How to run
+
+1. Set up your MySQL database with binary logging enabled
+2. Configure environment variables for MySQL connection
+3. Install dependencies: `pip install -r requirements.txt`
+4. Run: `python main.py`
+
+## Environment variables
+
+The connector uses the following environment variables:
+
+- **output**: Name of the output topic to write into.
+- **MYSQL_HOST**: The IP address or fully qualified domain name of your MySQL server.
+- **MYSQL_PORT**: The Port number to use for communication with the server (default: 3306).
+- **MYSQL_DATABASE**: The name of the database for CDC.
+- **MYSQL_USER**: The username that should be used to interact with the database.
+- **MYSQL_PASSWORD**: The password for the user configured above.
+- **MYSQL_SCHEMA**: The name of the schema/database for CDC (same as MYSQL_DATABASE).
+- **MYSQL_TABLE**: The name of the table for CDC.
+
+## Requirements / Prerequisites
+
+- A MySQL Database with binary logging enabled.
+- Set `log-bin=mysql-bin` and `binlog-format=ROW` in MySQL configuration.
+- MySQL user with `REPLICATION SLAVE` and `REPLICATION CLIENT` privileges.
+
+## Contribute
+
+Submit forked projects to the Quix [GitHub](https://github.com/quixio/quix-samples) repo. Any new project that we accept will be attributed to you and you'll receive $200 in Quix credit.
+
+## Open source
+
+This project is open source under the Apache 2.0 license and available in our [GitHub](https://github.com/quixio/quix-samples) repo.
+
+Please star us and mention us on social to show your appreciation.

python/sources/mysql_cdc/README_MYSQL_CDC.md

@@ -0,0 +1,124 @@
+# MySQL CDC Setup
+
+This application implements MySQL CDC using MySQL binary log replication.
+
+## Prerequisites
+
+1. **MySQL Configuration**: Your MySQL server must have binary logging enabled with ROW format:
+   ```ini
+   # Add to MySQL configuration file (my.cnf or my.ini)
+   log-bin=mysql-bin
+   binlog-format=ROW
+   server-id=1
+   ```
+
+2. **MySQL User Permissions**: The MySQL user needs REPLICATION SLAVE and REPLICATION CLIENT privileges:
+   ```sql
+   GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'your_user'@'%';
+   GRANT SELECT ON your_database.your_table TO 'your_user'@'%';
+   FLUSH PRIVILEGES;
+   ```
+
+## Environment Variables
+
+Set the following environment variables:
+
+### MySQL Connection
+- `MYSQL_HOST` - MySQL server hostname (e.g., localhost)
+- `MYSQL_PORT` - MySQL server port (default: 3306)
+- `MYSQL_USER` - MySQL username
+- `MYSQL_PASSWORD` - MySQL password
+- `MYSQL_DATABASE` - MySQL database name
+- `MYSQL_SCHEMA` - MySQL database name (same as MYSQL_DATABASE)
+- `MYSQL_TABLE` - Table name to monitor for changes
+
+### Kafka Output (unchanged)
+- `output` - Kafka topic name for publishing changes
+
+## Example .env file
+
+```env
+# MySQL Connection
+MYSQL_HOST=localhost
+MYSQL_PORT=3306
+MYSQL_USER=replication_user
+MYSQL_PASSWORD=your_password
+MYSQL_DATABASE=your_database
+MYSQL_SCHEMA=your_database
+MYSQL_TABLE=your_table
+
+# Kafka Output
+output=cdc-changes-topic
+```
+
+## Dependencies
+
+Install the required Python packages:
+```bash
+pip install -r requirements.txt
+```
+
+The key MySQL-specific dependencies are:
+- `pymysql` - MySQL database connector
+- `mysql-replication` - MySQL binary log replication library
+
+## Change Data Format
+
+The MySQL CDC produces change events in the following format:
+
+### INSERT Event
+```json
+{
+  "kind": "insert",
+  "schema": "database_name",
+  "table": "table_name", 
+  "columnnames": ["col1", "col2"],
+  "columnvalues": ["value1", "value2"],
+  "oldkeys": {}
+}
+```
+
+### UPDATE Event
+```json
+{
+  "kind": "update",
+  "schema": "database_name", 
+  "table": "table_name",
+  "columnnames": ["col1", "col2"],
+  "columnvalues": ["new_value1", "new_value2"],
+  "oldkeys": {
+    "keynames": ["col1", "col2"],
+    "keyvalues": ["old_value1", "old_value2"]
+  }
+}
+```
+
+### DELETE Event
+```json
+{
+  "kind": "delete",
+  "schema": "database_name",
+  "table": "table_name", 
+  "columnnames": [],
+  "columnvalues": [],
+  "oldkeys": {
+    "keynames": ["col1", "col2"],
+    "keyvalues": ["deleted_value1", "deleted_value2"]
+  }
+}
+```
+
+## Running the Application
+
+1. Ensure MySQL is configured with binary logging
+2. Set environment variables
+3. Run the application:
+   ```bash
+   python main.py
+   ```
+
+The application will:
+1. Connect to MySQL and validate binary logging is enabled
+2. Create a binary log stream reader
+3. Monitor the specified table for changes
+4. Buffer changes and publish them to Kafka every 500ms 

python/sources/mysql_cdc/app.yaml

@@ -0,0 +1,41 @@
+name: video-static-feature
+language: Python
+variables:
+  - name: output
+    inputType: OutputTopic
+    description: This is the output topic
+    defaultValue: none
+    required: true
+  - name: MYSQL_HOST
+    inputType: FreeText
+    description: Host name of MySQL
+    required: true
+  - name: MYSQL_PORT
+    inputType: FreeText
+    description: Port of MySQL
+    defaultValue: 3306
+    required: true
+  - name: MYSQL_USER
+    inputType: FreeText
+    description: Username of MySQL
+    required: true
+  - name: MYSQL_PASSWORD
+    inputType: HiddenText
+    description: Password of MySQL
+    required: true
+  - name: MYSQL_DATABASE
+    inputType: FreeText
+    description: Database name of MySQL
+    required: true
+  - name: MYSQL_SCHEMA
+    inputType: FreeText
+    description: Name of schema/database for CDC
+    required: true
+  - name: MYSQL_TABLE
+    inputType: FreeText
+    description: Name of table for CDC
+    required: true
+dockerfile: dockerfile
+runEntryPoint: main.py
+defaultFile: main.py
+libraryItemId: mysql-cdc-source

python/sources/mysql_cdc/dockerfile

@@ -0,0 +1,28 @@
+FROM python:3.12.5-slim-bookworm
+			
+# Set environment variables for non-interactive setup and unbuffered output
+ENV DEBIAN_FRONTEND=noninteractive \
+    PYTHONUNBUFFERED=1 \
+    PYTHONIOENCODING=UTF-8 \
+    PYTHONPATH="/app"
+			
+# Build argument for setting the main app path
+ARG MAINAPPPATH=.
+			
+# Set working directory inside the container
+WORKDIR /app
+			
+# Copy requirements to leverage Docker cache
+COPY "${MAINAPPPATH}/requirements.txt" "${MAINAPPPATH}/requirements.txt"
+			
+# Install dependencies without caching
+RUN pip install --no-cache-dir -r "${MAINAPPPATH}/requirements.txt"
+			
+# Copy entire application into container
+COPY . .
+			
+# Set working directory to main app path
+WORKDIR "/app/${MAINAPPPATH}"
+			
+# Define the container's startup command
+ENTRYPOINT ["python3", "main.py"]

python/sources/mysql_cdc/helper_functions.py

@@ -0,0 +1,101 @@
+from datetime import timedelta
+import os
+import json
+
+
+def load_config():
+    driver = os.environ["driver"]
+    server = os.environ["server"]
+    user_id = os.environ["userid"]
+    password = os.environ["password"]
+    database = os.environ["database"]
+    table_name = os.environ["table_name"]
+    last_modified_column = os.environ["last_modified_column"]
+    time_delta_config = os.environ["time_delta"]
+    
+    try:
+        use_utc_for_offset = bool(os.environ["offset_is_utc"])
+    except Exception as e:
+        raise Exception("Use UTC For Offset must be True or False", e)
+
+    drop_cols = os.getenv("columns_to_drop")
+    rename_cols = None
+    passed_rename_cols = os.getenv("columns_to_rename")
+    
+    try:
+        poll_interval = int(os.environ["poll_interval_seconds"])
+    except Exception as e:
+        raise Exception("Poll Interval must be an integer", e)
+
+    if poll_interval < 1:
+        poll_interval = 1
+
+    try:
+        if passed_rename_cols != None and passed_rename_cols != "":
+            rename_cols = json.loads(passed_rename_cols)
+    except Exception as e:
+        raise Exception("Invalid JSON supplied for column renames", e)
+
+    return {
+            "driver": driver,
+            "server": server,
+            "user_id": user_id,
+            "password": password,
+            "database": database,
+            "table_name": table_name,
+            "last_modified_column": last_modified_column,
+            "time_delta": make_time_delta_from_config(time_delta_config),
+            "drop_cols": drop_cols,
+            "rename_cols": rename_cols,
+            "use_utc": use_utc_for_offset,
+            "poll_interval": poll_interval
+            }
+
+
+def make_time_delta_from_config(time_delta_config) -> timedelta:
+    time_delta_values = time_delta_config.split(",")
+
+    if len(time_delta_values) != 5:
+        raise Exception(
+            "time_delta_config must contain 5 values, one for each of seconds, minutes, hours, days and weeks")
+
+    try:
+        seconds = int(time_delta_values[0])
+        minutes = int(time_delta_values[1])
+        hours = int(time_delta_values[2])
+        days = int(time_delta_values[3])
+        weeks = int(time_delta_values[4])
+        return timedelta(seconds = seconds, minutes = minutes, hours = hours, days = days, weeks = weeks)
+    except TypeError as te:
+        raise Exception("Unable to cast one of the supplied values to int", te)
+    except Exception as e:
+        raise Exception("Something went wrong configuring the time delta", e)
+
+
+def check_table_exists(conn, table) -> bool:
+    if not conn.cursor().tables(table).fetchone():
+        print("Table does not exist")
+        return False
+    return True
+
+
+def check_column_exists(conn, table, column_name) -> bool:
+    for c in conn.cursor().columns(table = table):
+        if column_name == c.column_name:
+            return True
+    print("Key column [{}] not found in table [{}]".format(column_name, table))
+    return False
+
+
+def drop_columns(conn, cols_to_drop, table_data, table_name) -> any:
+    for col in cols_to_drop:
+        if check_column_exists(conn, table_name, col):
+            table_data = table_data.drop(col, 1)
+    return table_data
+
+
+def rename_columns(conn, cols_to_rename, table_data, table_name) -> any:
+    for col in cols_to_rename:
+        if check_column_exists(conn, table_name, col):
+            table_data = table_data.rename(columns={col: cols_to_rename[col]})
+    return table_data