scala写flink消费kafka后数据写入hive和doris

首先，你需要在 Scala 代码中引入以下依赖：

libraryDependencies += "org.apache.flink" %% "flink-scala" % flinkVersion
libraryDependencies += "org.apache.flink" %% "flink-streaming-scala" % flinkVersion
libraryDependencies += "org.apache.flink" %% "flink-connector-kafka" % flinkVersion
libraryDependencies += "org.apache.flink" %% "flink-connector-hive" % flinkVersion
libraryDependencies += "org.apache.flink" %% "flink-connector-jdbc" % flinkVersion

然后，你可以使用以下代码来消费 Kafka 数据：

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

val env = StreamExecutionEnvironment.getExecutionEnvironment

val kafkaConsumer = new FlinkKafkaConsumer[String]("topic", new SimpleStringSchema(), properties)
val stream = env.addSource(kafkaConsumer)

// 对数据进行处理
val result = stream.map(...)

其中，`properties` 是一个 `Properties` 对象，用于配置 Kafka 的连接信息。

接下来，你需要将处理后的数据写入到 Hive 和 Doris 中。可以使用以下代码：

import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.table.catalog.hive.HiveCatalog
import org.apache.flink.streaming.api.scala.StreamTableEnvironment

val tableEnv = StreamTableEnvironment.create(env)

val hiveCatalog = new HiveCatalog("myHiveCatalog", "myDatabase", "/path/to/hive/conf", "2.3.4")
tableEnv.registerCatalog("myHiveCatalog", hiveCatalog)
tableEnv.useCatalog("myHiveCatalog")

tableEnv.executeSql("CREATE TABLE myHiveTable (...) WITH (...)")
result.toTable(tableEnv, "myResultTable")
tableEnv.executeSql("INSERT INTO myHiveTable SELECT * FROM myResultTable")

val jdbcUrl = "jdbc:mysql://localhost:3306/my_database"
tableEnv.executeSql(s"CREATE TABLE myDorisTable (...) WITH (...)")
tableEnv.executeSql(s"INSERT INTO myDorisTable SELECT * FROM myResultTable")

其中，`myHiveCatalog` 是 Hive 的 Catalog 名称，`myDatabase` 是 Hive 中的数据库名称，`/path/to/hive/conf` 是 Hive 的配置文件所在路径，`2.3.4` 是 Hive 的版本号。

`myHiveTable` 和 `myDorisTable` 是你要写入数据的表名，`(...)` 是表的列定义和其他属性，`myResultTable` 是处理后的数据表名。

`jdbcUrl` 是 Doris 数据库的连接信息，你需要根据实际情况进行修改。

你需要将上述代码中的 `...` 替换为实际的处理逻辑和表定义。