Shulou(Shulou.com)06/01 Report--

This article mainly shows you "how flink converts bathch dataset into sql operation", the content is simple and easy to understand, clear organization, I hope to help you solve doubts, let Xiaobian lead you to study and learn "how flink converts bathch dataset into sql operation" this article bar.

The sample code is as follows:

package flink.table_sql/* maven dependency required * org.apache.flink flink-table_2.12 1.7.1 */import org.apache.flink.api.scala._ import org.apache.flink.table.api.TableEnvironmentimport org.apache.flink.table.api.scala._ object FlinkTableTest { def main(args: Array[String]): Unit = { val env = ExecutionEnvironment.getExecutionEnvironment val persondata = env.readTextFile("d://person") val maps = persondata.map(str=>{ val strs = str.split(" ") person(strs(0).toInt,strs(1),strs(2).toInt) } ) val tableEnv = TableEnvironment.getTableEnvironment(env) //'id, 'name,'age is the mapped field, because the implicit conversion is used, the third line import will affect this tableEnv.registerDataSet("person", maps,'id, 'name,'age) val result = tableEnv.sqlQuery("select * from person where id = 1") val rows = result.collect() print(rows) env.execute("table_task") }}case class person(id:Int,name:String,age:Int) The above is "how to convert bathch dataset to sql operation" All the contents of this article, thank you for reading! I believe that everyone has a certain understanding, hope to share the content to help everyone, if you still want to learn more knowledge, welcome to pay attention to the industry information channel!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.