diff --git a/flow/connectors/mysql/cdc.go b/flow/connectors/mysql/cdc.go
index 49149baf5..1237071ce 100644
--- a/flow/connectors/mysql/cdc.go
+++ b/flow/connectors/mysql/cdc.go
@@ -241,7 +241,7 @@ func (c *MySqlConnector) SetupReplConn(context.Context, map[string]string) error
 	return nil
 }
 
-func (c *MySqlConnector) startSyncer(ctx context.Context) (*replication.BinlogSyncer, error) {
+func (c *MySqlConnector) startSyncer(ctx context.Context, env map[string]string) (*replication.BinlogSyncer, error) {
 	var tlsConfig *tls.Config
 	if !c.config.DisableTls {
 		var err error
@@ -271,6 +271,10 @@ func (c *MySqlConnector) startSyncer(ctx context.Context) (*replication.BinlogSy
 		config.Password = token
 	}
 
+	eventCacheCount, err := internal.PeerDBMySQLEventCacheCount(ctx, env)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get event cache count: %w", err)
+	}
 	//nolint:gosec
 	return replication.NewBinlogSyncer(replication.BinlogSyncerConfig{
 		ServerID:         rand.Uint32(),
@@ -286,12 +290,14 @@ func (c *MySqlConnector) startSyncer(ctx context.Context) (*replication.BinlogSy
 		ParseTime:        true,
 		TLSConfig:        tlsConfig,
 		HeartbeatPeriod:  c.binlogHeartbeatPeriod,
+		EventCacheCount:  eventCacheCount,
 	}), nil
 }
 
 func (c *MySqlConnector) startStreaming(
 	ctx context.Context,
 	pos string,
+	env map[string]string,
 ) (*replication.BinlogSyncer, *replication.BinlogStreamer, mysql.GTIDSet, mysql.Position, error) {
 	parsedOffset, err := parseReplicationOffsetText(c.Flavor(), pos)
 	if err != nil {
@@ -300,9 +306,9 @@ func (c *MySqlConnector) startStreaming(
 
 	switch parsedOffset.mechanism {
 	case protos.MySqlReplicationMechanism_MYSQL_FILEPOS.String():
-		return c.startCdcStreamingFilePos(ctx, parsedOffset.pos)
+		return c.startCdcStreamingFilePos(ctx, parsedOffset.pos, env)
 	case protos.MySqlReplicationMechanism_MYSQL_GTID.String():
-		return c.startCdcStreamingGtid(ctx, parsedOffset.gset)
+		return c.startCdcStreamingGtid(ctx, parsedOffset.gset, env)
 	default:
 		return nil, nil, nil, mysql.Position{}, fmt.Errorf("empty mysql replication offset")
 	}
@@ -311,8 +317,9 @@ func (c *MySqlConnector) startStreaming(
 func (c *MySqlConnector) startCdcStreamingFilePos(
 	ctx context.Context,
 	pos mysql.Position,
+	env map[string]string,
 ) (*replication.BinlogSyncer, *replication.BinlogStreamer, mysql.GTIDSet, mysql.Position, error) {
-	syncer, err := c.startSyncer(ctx)
+	syncer, err := c.startSyncer(ctx, env)
 	if err != nil {
 		return nil, nil, nil, mysql.Position{}, err
 	}
@@ -327,8 +334,9 @@ func (c *MySqlConnector) startCdcStreamingFilePos(
 func (c *MySqlConnector) startCdcStreamingGtid(
 	ctx context.Context,
 	gset mysql.GTIDSet,
+	env map[string]string,
 ) (*replication.BinlogSyncer, *replication.BinlogStreamer, mysql.GTIDSet, mysql.Position, error) {
-	syncer, err := c.startSyncer(ctx)
+	syncer, err := c.startSyncer(ctx, env)
 	if err != nil {
 		return nil, nil, nil, mysql.Position{}, err
 	}
@@ -389,7 +397,7 @@ func (c *MySqlConnector) PullRecords(
 		return fmt.Errorf("failed to determine if binlog row metadata is supported: %w", err)
 	}
 
-	syncer, mystream, gset, pos, err := c.startStreaming(ctx, req.LastOffset.Text)
+	syncer, mystream, gset, pos, err := c.startStreaming(ctx, req.LastOffset.Text, req.Env)
 	if err != nil {
 		return err
 	}
diff --git a/flow/connectors/mysql/ssh_keepalive_test.go b/flow/connectors/mysql/ssh_keepalive_test.go
index 76183e4f5..fb8929cea 100644
--- a/flow/connectors/mysql/ssh_keepalive_test.go
+++ b/flow/connectors/mysql/ssh_keepalive_test.go
@@ -316,7 +316,7 @@ func TestMySQLCloseSyncerWithTimeout(t *testing.T) {
 
 	pos, err := connector.GetMasterPos(ctx)
 	require.NoError(t, err)
-	syncer, _, _, _, err := connector.startCdcStreamingFilePos(ctx, pos) //nolint:dogsled
+	syncer, _, _, _, err := connector.startCdcStreamingFilePos(ctx, pos, nil) //nolint:dogsled
 	require.NoError(t, err)
 
 	// Let CDC streaming establish before blocking the MySQL server.
diff --git a/flow/connectors/mysql/ssh_test.go b/flow/connectors/mysql/ssh_test.go
index 811e4c07e..755339f91 100644
--- a/flow/connectors/mysql/ssh_test.go
+++ b/flow/connectors/mysql/ssh_test.go
@@ -53,7 +53,7 @@ func TestMySQLSyncerClose(t *testing.T) {
 
 	pos, err := connector.GetMasterPos(ctx)
 	require.NoError(t, err)
-	syncer, _, _, _, err := connector.startCdcStreamingFilePos(ctx, pos) //nolint:dogsled
+	syncer, _, _, _, err := connector.startCdcStreamingFilePos(ctx, pos, nil) //nolint:dogsled
 	require.NoError(t, err)
 
 	// Let CDC streaming establish before closing the syncer.
diff --git a/flow/internal/dynamicconf.go b/flow/internal/dynamicconf.go
index 143c54b8a..21c4fb47b 100644
--- a/flow/internal/dynamicconf.go
+++ b/flow/internal/dynamicconf.go
@@ -460,6 +460,14 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_AFTER_RESUME,
 		TargetForSetting: protos.DynconfTarget_POSTGRES,
 	},
+	{
+		Name:             "PEERDB_MYSQL_EVENT_CACHE_COUNT",
+		Description:      "Maximum number of events the MySQL Go driver loads in memory at once, useful when pushing large rows",
+		DefaultValue:     "10240",
+		ValueType:        protos.DynconfValueType_INT,
+		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_IMMEDIATE,
+		TargetForSetting: protos.DynconfTarget_ALL,
+	},
 }
 
 var DynamicIndex = func() map[string]int {
@@ -613,6 +621,10 @@ func PeerDBCDCChannelBufferSize(ctx context.Context, env map[string]string) (int
 	return dynamicConfSigned[int](ctx, env, "PEERDB_CDC_CHANNEL_BUFFER_SIZE")
 }
 
+func PeerDBMySQLEventCacheCount(ctx context.Context, env map[string]string) (int, error) {
+	return dynamicConfSigned[int](ctx, env, "PEERDB_MYSQL_EVENT_CACHE_COUNT")
+}
+
 func PeerDBNormalizeBufferHours(ctx context.Context, env map[string]string) (int64, error) {
 	return dynamicConfSigned[int64](ctx, env, "PEERDB_NORMALIZE_BUFFER_HOURS")
 }