scala 对相同key下 ，相同的value只保留一条

在Scala中，可以使用`groupBy`和`mapValues`方法来实现对相同key下，相同的value只保留一条的操作。具体实现方法如下：

val data = List(("key1", "value1"), ("key1", "value1"), ("key1", "value2"), ("key2", "value1"), ("key2", "value2"))

val result = data.groupBy(_._1) // 按key分组
  .mapValues(_.map(_._2).distinct) // 对每组的value去重

println(result) // 输出结果：Map(key1 -> List(value1, value2), key2 -> List(value1, value2))

在上面的代码中，首先使用`groupBy`方法按key对数据进行分组，得到一个Map类型的结果。然后使用`mapValues`方法对每组的value进行去重，得到最终的结果。

其中`_.map(_._2)`表示对每组数据中的元组进行映射，取出value；`.distinct`表示对取出的value进行去重操作。

相关问题

scala的flink+kafka实时来一条统计pv

使用Scala的Flink和Kafka实时来一条统计PV的方法如下：

首先，我们需要创建一个Flink的流处理任务。在任务中，我们可以使用Flink提供的Kafka Consumer来消费Kafka中的消息流，并使用Flink的处理函数对消息进行处理。

在处理函数中，我们可以将消费到的每条消息的PV字段进行累加。假设每条消息中包含一个PV字段（表示Page Views，即页面访问量），我们可以定义一个累加器，并使用Flink的MapState来保存当前的PV值。

下面是一个简单的示例代码：

import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.api.common.typeinfo.{TypeHint, TypeInformation}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

object PVStatistics {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val kafkaProps = new Properties()
    kafkaProps.setProperty("bootstrap.servers", "localhost:9092")
    kafkaProps.setProperty("group.id", "pv-consumer-group")

    val inputTopic = "pv-topic"

    val kafkaConsumer = new FlinkKafkaConsumer[String](inputTopic, new SimpleStringSchema(), kafkaProps)
    val stream = env.addSource(kafkaConsumer)

    // 定义累加器和MapStateDescriptor
    val pvAccumulator = new IntCounter
    val pvStateDescriptor = new MapStateDescriptor[String, Int]("pv-state", TypeInformation.of(new TypeHint[String]{}), TypeInformation.of(new TypeHint[Int]{}))

    val pvStream = stream.map(new MapFunction[String, Int] {
      override def map(value: String): Int = {
        pvAccumulator.add(1)
        pvAccumulator.getLocalValue
      }
    }).keyBy(_ => "pv-key")
      .mapWithState[(String, Int), MapState[String, Int]] {
        // 更新PV值并返回累加结果
        case (value, state: MapState[String, Int]) =>
          val pv = state.get("pv")
          val newPv = pv + value
          state.put("pv", newPv)
          ((inputTopic, newPv), state)
      }
    
    pvStream.print()

    env.execute("PV Statistics")
  }
}

在上述代码中，我们定义了一个`pvAccumulator`作为累加器，并通过`pvStateDescriptor`创建了一个MapState来保存每个topic的PV值。

然后，我们使用`FlinkKafkaConsumer`创建了一个Kafka Consumer，并从指定的topic `pv-topic`中消费消息流。接着，我们使用`map`函数将每一条消息的PV字段累加到累加器中，并将累加结果输出为`(topic, pv)`的元组形式。

最后，我们使用`execute`方法执行Flink任务，即开始实时统计PV。

以上是使用Scala的Flink和Kafka实时统计PV的一个简单示例。实际情况中，你可能需要根据具体需求进行更详细的配置和调整。

dataframe某一列保留两位小数 scala

在 Scala 中，可以使用 `round` 函数来保留 DataFrame 中某一列的小数位数。以下是一个示例代码：

import org.apache.spark.sql.functions._

val df = Seq((1, 2.345), (2, 3.456), (3, 4.567)).toDF("id", "value")

val dfWithRoundedValue = df.withColumn("rounded_value", round(col("value"), 2))

dfWithRoundedValue.show()

在上面的代码中，我们使用 `withColumn` 函数添加了一个名为 `rounded_value` 的新列，其中包含了 `value` 列的保留两位小数后的值。`round` 函数接受两个参数：第一个参数是要舍入的列，第二个参数是要保留的小数位数。

输出结果如下：

+---+-----+-------------+
| id|value|rounded_value|
+---+-----+-------------+
|  1|2.345|         2.35|
|  2|3.456|         3.46|
|  3|4.567|         4.57|
+---+-----+-------------+