Maxwell Documentation

Maxwell tracks database schema changes to interpret raw MySQL binlog bytes as typed data (numbers, strings, etc.). It uses a combination of base tables and a delta-based history system stored in the maxwell database.

Base Schema Tables

When Maxwell first runs, it captures the initial schema in these tables:

• tables
• columns
• databases

Schema Change History

As schema modifications occur in the binlog, Maxwell stores the changes (diffs) in the schemas table. Each entry in schemas contains:

• binlog_file, binlog_position (or gtid_set): The exact binlog location of the change.
• deltas: The internal representation of the schema change.
• base_schema_id: The ID of the previous schema this delta applies to.
• last_heartbeat_read: The most recent Maxwell heartbeat seen in the binlog before this change.
• server_id: The identifier for the database server.

To reconstruct the schema at any specific binlog position, Maxwell finds the most recent schema for the server_id that occurred before that position, then follows the base_schema_id chain back to the initial captured schema.

For complex filtering or data munging requirements, you can provide a Javascript file via the --javascript FILE flag.

Your script must contain a function named process_row(row, state).

Arguments:

• row: A WrappedRowMap object representing the current row. It provides methods like row.suppress() to drop the row and row.data.get(key) / row.data.put(key, value) to access/modify row data.
• state: A LinkedHashMap<String, Object> representing a global state that persists across rows. You can use state.put(key, value) and state.get(key) to maintain state for filtering decisions (e.g., tracking a flag across multiple rows).

Capabilities:

• Suppressing rows: Call row.suppress() to prevent the row from being emitted.
• Data Munging: Modify the row data directly using row.data.put() before it is emitted.

function process_row(row, state) {
	// Example: Updating global state based on row values
	if ( row.database == "test" && row.table == "lock") {
		var haslock = row.data.get("haslock");
		if ( haslock == "false" ) {
			state.put("haslock", "false");
		} else if( haslock == "true" ) {
			state.put("haslock", "true");
		}
	}

	// Example: Suppressing rows based on the global state
	if(state.get("haslock") == "true") {
		row.suppress();
	}

	// Example: Filtering and modifying data based on actual values
	if ( row.database == "test" && row.table == "bar" ) {
		var username = row.data.get("username");
		if ( username == "osheroff" )
			row.suppress();

		row.data.put("username", username.toUpperCase());
	}
}

Maxwell supports partitioning for Kafka, AWS Kinesis, and SNS/SQS. You can control how data is distributed using the producer_partition_by option.

Available partitioning strategies:

• database
• table
• primary_key
• transaction_id
• column_data
• random

Partitioning by Column Data: If you choose column_data, you must provide:

1. producer_partition_columns: A comma-separated list of column names.
2. producer_partiton_by_fallback: A fallback strategy (_database_, _table_, or _primary_key_) to use if the specified column does not exist in a row.

Kafka Specifics: Kafka partitions are determined by HASH_FUNCTION(producer_partion_value) % TOPIC.NUMBER_OF_PARTITIONS.

• The default HASH_FUNCTION is hashCode.
• You can set kafka_partition_hash to murmurhash3 (seed is hardcoded to 25342).
• Important: You should pre-create your Kafka topics with the desired number of partitions before starting Maxwell.