Spark-Java：如何更改 Dataset<Row> 中列的时间戳格式？

Question

我想将我的时间戳字段映射到 Dataset 中，其值2018-08-17T19:58:46.000+0000的格式类似于yyyy-MM-dd HH:mm:ss.SSS2018-08-17 19:58:46.000，以及一些列到yyyy-MM-dd。例如，我有一个数据集DS1，其列id、lastModif、created：+------------------+----------------------------+----------------------------+|Id                |lastModif                   |created                     |+------------------+----------------------------+----------------------------+|abc1              |2019-01-14T19:51:55.000+0000|2019-02-07T20:37:53.000+0000||AQA2              |2019-02-05T19:26:36.000+0000|2019-02-07T20:40:06.000+0000|+------------------+----------------------------+----------------------------+ 从上面的 DS1 我需要lastModif映射到格式的列yyyy-MM-dd HH:mm:ss.SSS和createdTime映射到的列yyyy-MM-dd。我有类似的 DS2，DS3 具有不同的列映射。我保留了一个属性文件，它将从中获取映射列作为键和时间戳格式作为值。在代码中，我保留了映射列和非映射列的列表，并选择了该列：String cols = "Id,created,lastModif";String[] colArr = cols.split(",");String mappedCols = "lastModif,created"; //hardcoding as of now.List<String> mappedColList = Arrays.asList(mappedCols.split(","));String nonMappedCols = getNonMappingCols(colArr, mappedCols.split(",")).toLowerCase();List<String> nonMapped = Arrays.asList(nonMappedCols.split(","));//column-mapping logicfiltered = tempDS.selectExpr(convertListToSeq(nonMapped),unix_timestamp($"lastModif","yyyy-MM-dd HH:mm:ss.SSS").cast("timestamp").as("lastModif"));filtered.show(false);public static Seq<String> convertListToSeq(List<String> inputList){    return JavaConverters.asScalaIteratorConverter(inputList.iterator()).asScala().toSeq();}如何将列映射到所需的时间戳格式？并且在代码行中tempDS.selectExpr(convertListToSeq(nonMapped),unix_timestamp($"lastModif","yyyy-MM-dd HH:mm:ss.SSS").cast("timestamp").as("lastModif"));，$"lastModif"Java 中无法识别。其次，这种方式是一种静态方式，即对映射列进行硬编码。如何映射我的列List<String> mappedColList？

Answer 1

首先，让我们澄清您的输入数据。您提到过，您有Timestamp，但您列出的输出格式似乎只是字符串值，表示以下格式yyyy-MM-dd'T'HH:mm:ss.SSSZ。你能证实这个结论吗？

在您的一条评论中，您回答该函数在您的尝试中unix_timestamp返回。null查看unix_timestamp(Column s, String p)的文档，我们可以看到它需要其他格式来解析，否则它会返回null：

参数：

s - 日期、时间戳或字符串。如果是字符串，则数据必须采用可以转换为时间戳的格式，例如yyyy-MM-dd或yyyy-MM-dd HH:mm:ss.SSSS

fmt - 详细说明 s 格式的日期时间模式当 s 是字符串时

返回：

时间戳，如果 s 是无法转换为时间戳的字符串或 fmt 是无效格式，则返回null

如果您输入的参数确实是字符串，我建议您使用以下解决方案，使用 Spark SQL 函数to_timestamp(Column s, String fmt)和date_format(Column dateExpr, String format)

import static org.apache.spark.sql.functions.col;

import static org.apache.spark.sql.functions.to_timestamp;

import static org.apache.spark.sql.functions.date_format;

....

SparkSession spark = SparkSession

            .builder()

            .appName("datetime-transformation")

            .master("local[*]")

            .getOrCreate();

SomeDto someDto = SomeDto.builder()

            .id("abc1")

            .lastModif("2019-01-14T19:51:55.123+02:00")

            .created("2019-01-14T19:51:55.123+02:00")

            .build();

Dataset<Row> ds = spark.createDataset(Collections.singletonList(someDto), Encoders.bean(SomeDto.class)).toDF();

ds.printSchema();

ds.show(false);

Dataset<Row> dfm = ds

            .withColumn("lastModif", to_timestamp(col("lastModif"), "yyyy-MM-dd'T'HH:mm:ss.SSSXXX"))

            .withColumn("created", date_format(to_timestamp(col("lastModif"), "yyyy-MM-dd'T'HH:mm:ss.SSSXXX"), "yyyy-MM-dd"));

dfm.printSchema();

dfm.show(false);

输出将是：

root

 |-- created: string (nullable = true)

 |-- id: string (nullable = true)

 |-- lastModif: string (nullable = true)

+-----------------------------+----+-----------------------------+

|created                      |id  |lastModif                    |

+-----------------------------+----+-----------------------------+

|2019-01-14T19:51:55.123+02:00|abc1|2019-01-14T19:51:55.123+02:00|

+-----------------------------+----+-----------------------------+

root

 |-- created: string (nullable = true)

 |-- id: string (nullable = true)

 |-- lastModif: timestamp (nullable = true)

+----------+----+-----------------------+

|created   |id  |lastModif              |

+----------+----+-----------------------+

|2019-01-14|abc1|2019-01-14 19:51:55.123|

+----------+----+-----------------------+

Answer 2

这就是我使映射动态化的方式：

private static Dataset<Row> mapColumns(Properties mappings, String tableNm, String[] colArr, Dataset<Row> tempDS) throws Exception

{

    String mappedCols = "lastmodif,createdDate,endDate";

    Dataset<Row> filtered = null;

    Properties mappingCols = mappings;

    List<String> mapped = Arrays.asList(mappedCols.split(","));

    List<String> colsList = Arrays.asList(colArr);

    ArrayList<String> tempList = new ArrayList<String>();

    Iterator itrTmp = colsList.iterator();

    while(itrTmp.hasNext()){

        tempList.add((String)itrTmp.next());

    }

    Iterator itr = mapped.iterator();

    filtered = tempDS.selectExpr(convertListToSeq(colsList));

    while(itr.hasNext()){

        String column = itr.next().toString();

        String newCol = column+"_mapped";

        String propertyKey = tableNm+"-"+column;

        String propertyValue = mappingCols.getProperty(propertyKey);

        filtered = filtered.selectExpr(convertListToSeq(colsList))

                .withColumn(newCol, functions.regexp_replace(functions.substring(filtered.col(column), 0, 23),"T", " ")).alias(newCol)

                .drop(filtered.col(column));

        tempList.remove(column);

        tempList.add(newCol);

        colsList = tempList;

    }

    filtered = filtered.selectExpr(convertListToSeq(colsList)); 

    filtered.show(false);

}

public static Seq<String> convertListToSeq(List<String> inputList)

{

    return JavaConverters.asScalaIteratorConverter(inputList.iterator()).asScala().toSeq();

}

但是StringtoTimestamp转换仍然悬而未决。截至目前，我正在做一个substring, 但是这个逻辑适用于所有数据类型为yyyy-mm-ddThh:mm:ss.SSSZ等yyyy-mm-ddThh:mm:ss.SSS+0000的列，但如果列具有类型yyyy-mm-dd的数据并且代码将中断，则该逻辑将不起作用。我在这里提出了这个：如何将字符串转换为时间戳。