MySQL是一種常見的關系型數據庫，而Flink是流處理引擎，兩者的組合可以實現MySQL數據實時落到Flink上，以便進行實時的處理和分析。

要將MySQL的數據實時落到Flink上，需要借助Flink提供的JDBC數據源，具體步驟如下：

Properties properties = new Properties();
properties.setProperty("driver","com.mysql.jdbc.Driver");
properties.setProperty("url","jdbc:mysql://localhost:3306/test");
properties.setProperty("user","root");
properties.setProperty("password","123456");
DataStream<Tuple2<String, Integer>> dataStream = env.addSource(
new JdbcSourceFunction<>(
properties,
"SELECT name, age FROM user WHERE age >18",
new TupleRowConverter()
)
);

在代碼示例中，定義了一個Properties對象，包含了MySQL的連接信息以及賬戶密碼等，并通過JdbcSourceFunction將MySQL中的user表數據拉取到Flink中。JdbcSourceFunction需要傳入三個參數：連接信息、SQL語句以及TupleRowConverter對象，后者用于將查詢結果轉為Tuple2<String, Integer>形式的DataStream。需要注意的是，SQL語句中可以加入條件語句，以過濾不必要的數據。

接下來，可以針對DataStream進行各種實時的計算和操作，比如使用Map函數進行字段轉換、使用Filter函數進行條件過濾、使用KeyBy函數進行分組等。Flink的操作和流程可參考官方文檔。

最后，在將結果落地到MySQL之前，需要先將結果轉換為符合MySQL表結構的形式。一個簡單的示例代碼如下：

dataStream.addSink(
new JdbcOutputFormat<>(
"INSERT INTO result(name, age) VALUES(?, ?)",
new JDBCParameterSetter<>() {
public void setParameter(PreparedStatement preparedStatement, Tuple2<String, Integer> tuple2) throws SQLException {
preparedStatement.setString(1, tuple2.f0);
preparedStatement.setInt(2, tuple2.f1);
}
},
new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
.withUrl("jdbc:mysql://localhost:3306/test")
.withDriverName("com.mysql.jdbc.Driver")
.withUsername("root")
.withPassword("123456")
.build()
)
);

代碼中使用JDBCParameterSetter將Tuple2轉化為符合MySQL表結構的形式，然后通過JdbcOutputFormat將結果插入到名為result的表中，最終實現了MySQL數據實時落到Flink上的功能。