scala的list添加timestamp类型数据

可以使用 scala 的 ListBuffer 类型来添加 timestamp 类型数据，然后使用 map 函数将 timestamp 类型转换为字符串类型。以下是一个示例代码：

import scala.collection.mutable.ListBuffer
import java.sql.Timestamp

val list = ListBuffer[Timestamp]()

val timestamp = new Timestamp(System.currentTimeMillis())
list += timestamp

val stringList = list.map(_.toString)
println(stringList)

这将在控制台输出一个字符串形式的 timestamp 类型数据。

相关问题

scala的list添加timestamp、String类型数据

可以通过定义一个包含 timestamp 和 String 的数据结构来实现。例如：

case class MyData(timestamp: Long, data: String)

然后，可以定义一个 List[MyData] 类型的列表，并在需要添加数据时创建一个包含时间戳和字符串数据的 MyData 对象，将其添加到 List 中即可。例子如下：

val myList = List[MyData]()
val newData = MyData(System.currentTimeMillis(), "Some String Data")
val updatedList = newData :: myList

注意，以上代码只是一个示例，实际使用时需要适应具体的场景和需求。

flink 实现mysql cdc scala 代码

以下是使用 Flink 实现 MySQL CDC 的 Scala 代码示例：

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.Collector
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.types.Row
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema
import org.apache.flink.streaming.connectors.kafka.KafkaContextAware
import org.apache.flink.streaming.connectors.kafka.KafkaContextAware.KafkaContext
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWrapper
import java.util.Properties

object MySQLCDC {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    properties.setProperty("group.id", "flink-group")

    val consumer = new FlinkKafkaConsumer011[String]("mysql-cdc", new SimpleStringSchema(), properties)
    val stream = env.addSource(consumer).map(new MapFunction[String, Row]() {
      override def map(value: String): Row = {
        val fields = value.split(",")
        Row.of(fields(0).toInt.asInstanceOf[Object], fields(1).asInstanceOf[Object], fields(2).asInstanceOf[Object])
      }
    }).assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks[Row]() {
      override def extractTimestamp(row: Row, previousTimestamp: Long): Long = {
        row.getField(0).asInstanceOf[Int].toLong
      }

      override def checkAndGetNextWatermark(row: Row, extractedTimestamp: Long): Watermark = {
        new Watermark(extractedTimestamp)
      }
    })

    val windowedStream = stream.keyBy(1).timeWindow(Time.seconds(10)).apply(new WindowFunction[Row, Row, Tuple, TimeWindow]() {
      override def apply(key: Tuple, window: TimeWindow, input: Iterable[Row], out: Collector[Row]): Unit = {
        val sortedInput = input.toList.sortBy(_.getField(0).asInstanceOf[Int])
        val firstRow = sortedInput.head
        val lastRow = sortedInput.last
        out.collect(Row.of(firstRow.getField(1), firstRow.getField(2), lastRow.getField(2)))
      }
    })

    val producer = new FlinkKafkaProducer011[String]("mysql-cdc-output", new KafkaSerializationSchema[String]() with KafkaContextAware[String] {
      var context: KafkaContext = _

      override def serialize(element: String, timestamp: java.lang.Long): org.apache.kafka.clients.producer.ProducerRecord[Array[Byte], Array[Byte]] = {
        new org.apache.kafka.clients.producer.ProducerRecord(context.getOutputTopic(), element.getBytes())
      }

      override def setRuntimeContext(context: KafkaContext): Unit = {
        this.context = context
      }
    }, properties, FlinkKafkaProducer011.Semantic.EXACTLY_ONCE)

    windowedStream.map(new MapFunction[Row, String]() {
      override def map(row: Row): String = {
        s"${row.getField(0)},${row.getField(1)},${row.getField(2)}"
      }
    }).addSink(producer)

    env.execute("MySQL CDC")
  }
}