Spark Dataframe 为浮点数提供不同级别的精度

Posted 2023-04-15

【中文标题】Spark Dataframe 为浮点数提供不同级别的精度

【英文标题】：Spark Dataframe giving different levels of precision for floats

【发布时间】：2020-02-06 20:26:12

【问题描述】：

当我们创建一个 spark 数据帧时，我们将数据帧中的数据发送到 Kudu 和 Kafka（依次被拾取并进入 S3）

但是，一旦数据帧被更新，我们在 Kudu 中看到的精度是 1e-15

现在，如果我使用相同的数据帧并将其转换为 Kafka 可以使用的格式

override def getKafkaDataFrame(df: DataFrame) : DataFrame = 
df.withColumn("key", to_json(struct(df.col(PK1),
  df.col(PK2)))
  .withColumn("value", to_json(struct(df.columns.map(col): _*)))

然后（在单独的应用程序中）发送到 S3

对于 1e-6 的同一行，结果是这样的

我们还有一种方法可以直接从 Kudu 到 S3（不是流），当 inferSchema 设置为 true 时，精度与 Kudu 匹配，然后是一个简单的 dataframe.write

我想知道这种精度/规模损失是从哪里来的，我该如何解决它。

尝试了什么： 将所有字段转换为字符串， 仅将浮点字段转换为双精度数

谢谢

【参考方案1】：

不确定数据框使用了哪个数据源，但很可能问题是由存在精度问题的列的架构和数据类型引起的。 FloatType 不足以处理 1e-15 精度，因为它表示 4 字节单精度浮点数。

小例子

object DecisionPlays 
  import org.apache.spark.sql.SparkSession
  import org.apache.spark.sql.functions._
  import org.apache.spark.sql.types.FloatType

  case class DoubleColumn(doubleValue:Double)

  def main(args: Array[String]): Unit = 
    val spark = SparkSession.builder().master("local[*]").getOrCreate()
    val temp = DoubleColumn(1.12345678910111213)
    val df = spark.createDataFrame(Seq(temp))
    //[1.1234567891011122,"doubleValue":1.1234567891011122]
    df.withColumn("value", to_json(struct(df.columns.map(col): _*))).collect().foreach(println)
    //[1.1234568,"doubleValue":1.1234568]
    df.select(col("doubleValue").cast(FloatType))
      .withColumn("value", to_json(struct(df.columns.map(col): _*))).collect().foreach(println)

  

【讨论】：

奇怪的是，当我执行 printSchema 时，它说该列是一个浮点数。很明显，当我们将它发送到 Kudu 时它并没有改变，这让我感到困惑，为什么我尝试将它转换为 double 它仍然是错误的

【参考方案2】：

以安德烈所说的火花漂浮物

我决定创建一个浮点型字段，并在代码中的任何位置将其更改为双精度

它成功了！

Kudu 已经将所有浮点数更改为双精度数，因此那里没有发生任何数据更改，但现在数据在 s3 中也是正确的