spark-sql 对空值处理

Spark SQL 在处理空值时可以使用以下函数：

1. `coalesce`: 如果参数中有非空值，则返回第一个非空值，否则返回 `null`。

2. `nvl` 和 `nvl2`: 如果第一个参数为 `null`，则返回第二个参数，否则返回第一个参数。`nvl2` 还有第三个参数，表示如果第一个参数不为 `null`，则返回第二个参数，否则返回第三个参数。

3. `ifnull` 和 `nullif`: `ifnull` 接收两个参数，如果第一个参数不为 `null`，则返回第一个参数，否则返回第二个参数。`nullif` 接收两个参数，如果两个参数相等，则返回 `null`，否则返回第一个参数。

4. `when` 和 `otherwise`: `when` 可以根据条件返回不同的值，`otherwise` 表示默认情况下的返回值。

5. `drop` 和 `fillna`: `drop` 可以删除包含 `null` 值的行或列，`fillna` 可以用指定的值或策略填充 `null` 值。

例如，使用 `coalesce` 函数处理空值：

SELECT coalesce(name, 'Unknown') FROM people;

如果 `name` 字段有值，则返回该值，否则返回 `'Unknown'`。

相关问题

spark SQL连续12行空值 则过滤

可以使用Spark SQL的窗口函数来实现连续12行空值的过滤。

以一个示例数据集为例：

| id | value |
|----|-------|
| 1 | null |
| 2 | 10 |
| 3 | null |
| 4 | null |
| 5 | 20 |
| 6 | null |
| 7 | null |
| 8 | null |
| 9 | null |
| 10 | null |
| 11 | null |
| 12 | null |
| 13 | 30 |
| 14 | null |
| 15 | null |

首先，我们需要为每一行添加一个序号，可以使用Spark SQL的内置函数`row_number()`实现：

SELECT ROW_NUMBER() OVER (ORDER BY id) AS row_num, id, value FROM my_table

接下来，我们可以使用窗口函数`LAG()`来获取前一行的值。如果前一行的值为null，并且当前行的值也为null，那么就可以认为这是连续的空值。因此，我们可以使用窗口函数`SUM()`来计算连续的空值数量，并且过滤掉连续空值数量大于等于12的行：

SELECT id, value
FROM (
  SELECT ROW_NUMBER() OVER (ORDER BY id) AS row_num, id, value,
         SUM(CASE WHEN value IS NULL AND LAG(value) OVER (ORDER BY id) IS NULL THEN 1 ELSE 0 END)
           OVER (ORDER BY id) AS null_count
  FROM my_table
) tmp
WHERE null_count < 12 OR null_count IS NULL
ORDER BY id

这样，就可以过滤掉连续12行或以上的空值了。

scala如何将DataFrame的Spark sql表达式中的空值写入数据库表

如果您想将DataFrame中的空值写入数据库表，可以使用Spark SQL中的`null`函数。具体步骤如下：

1. 首先，您需要将DataFrame注册为一个临时表，可以使用`createOrReplaceTempView`方法。

val df = // your DataFrame
df.createOrReplaceTempView("temp_table")

2. 接下来，使用`insertInto`方法将数据插入到数据库表中。使用`null`函数替换空值。

spark.sql("INSERT INTO my_table SELECT IFNULL(col1, null), IFNULL(col2, null) FROM temp_table")

在上面的代码中，`IFNULL`函数将空值替换为`null`，然后将结果插入到数据库表中。

请注意，`INSERT INTO`语句需要与数据库类型和驱动程序相匹配。如果您使用的是MySQL数据库，可以使用以下代码：

spark.sql("INSERT INTO my_table SELECT IFNULL(col1, null), IFNULL(col2, null) FROM temp_table")

如果您使用的是PostgreSQL数据库，可以使用以下代码：

spark.sql("INSERT INTO my_table SELECT COALESCE(col1, null), COALESCE(col2, null) FROM temp_table")

这是因为PostgreSQL使用`COALESCE`函数替换空值。