replace custom UDF with spark from_unixtime for ctime transformation

src/main/java/com/teragrep/pth10/ast/commands/transformstatement/convert/Ctime.java

@@ -45,33 +45,44 @@
  */
 package com.teragrep.pth10.ast.commands.transformstatement.convert;
 
-import com.teragrep.pth10.ast.DPLTimeFormat;
-import org.apache.spark.sql.api.java.UDF2;
+import org.apache.spark.sql.api.java.UDF1;
 
-import java.text.DateFormat;
-import java.util.Date;
-import java.util.TimeZone;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
 
 /**
  * UDF for convert command 'ctime'<br>
  * Converts epoch time into given timeformat<br>
  * 
  * @author eemhu
  */
-public class Ctime implements UDF2<String, String, String> {
+public class Ctime implements UDF1<String, String> {
 
     private static final long serialVersionUID = 1L;
+    private final DateTimeFormatter formatter;
 
-    @Override
-    public String call(String epoch, String tf) throws Exception {
-        Long e = Long.valueOf(epoch);
-
-        Date date = new Date(e * 1000L);
-        DateFormat format = new DPLTimeFormat(tf).createSimpleDateFormat();
-        format.setTimeZone(TimeZone.getTimeZone("Etc/UTC"));
-        String formatted = format.format(date);
+    public Ctime(final String format) {
+        this(DateTimeFormatter.ofPattern(format).withZone(ZoneId.of("UTC")));
+    }
 
-        return formatted;
+    public Ctime(final DateTimeFormatter formatter) {
+        this.formatter = formatter;
     }
 
+    @Override
+    public String call(String epoch) throws Exception {
+        String result;
+        try {
+            Instant instant = Instant.from(formatter.parse(epoch));
+            result = instant.toString();
+        }
+        catch (DateTimeParseException e) {
+            throw new IllegalArgumentException(
+                    "Could not parse value <" + epoch + "> with set time formatter <" + formatter + ">"
+            );
+        }
+        return result;
+    }
 }

src/main/java/com/teragrep/pth10/steps/convert/AbstractConvertStep.java

@@ -54,7 +54,7 @@ public abstract class AbstractConvertStep extends AbstractStep {
 
     protected List<ConvertCommand> listOfCommands = new ArrayList<>();
     protected List<String> listOfFieldsToOmit = new ArrayList<>();
-    protected String timeformat = "%m/%d/%Y %H:%M:%S";
+    protected String timeformat = "yyyy-MM-dd'T'HH:mm:ssX";
 
     public AbstractConvertStep() {
         super();

src/main/java/com/teragrep/pth10/steps/convert/ConvertStep.java

@@ -231,11 +231,10 @@ private Dataset<Row> mktime(Dataset<Row> dataset, String field, String renameFie
      * @return Input dataset with added result column
      */
     private Dataset<Row> ctime(Dataset<Row> dataset, String field, String renameField) {
-        UserDefinedFunction ctimeUDF = functions.udf(new Ctime(), DataTypes.StringType);
-        sparkSession.udf().register("UDF_Ctime", ctimeUDF);
-
-        Column udfResult = functions
-                .callUDF("UDF_Ctime", functions.col(field).cast(DataTypes.StringType), functions.lit(timeformat));
+        final Ctime ctime = new Ctime(timeformat);
+        final UserDefinedFunction cTimeUDF = functions.udf(ctime, DataTypes.StringType);
+        sparkSession.udf().register("UDF_Ctime", cTimeUDF);
+        final Column udfResult = functions.callUDF("UDF_Ctime", functions.col(field).cast(DataTypes.StringType));
         return dataset.withColumn(renameField == null ? field : renameField, udfResult);
     }
 

src/test/java/com/teragrep/pth10/ConvertTransformationTest.java

@@ -108,18 +108,18 @@ void testConvertCtimeAs() {
                     new StructField("new", DataTypes.StringType, true, new MetadataBuilder().build())
             });
             Assertions.assertEquals(expectedSchema, ds.schema());
-
+            // match yyyy-MM-dd'T'HH:mm:ssZ ISO 8601 with zone
+            Pattern iso8601pattern = Pattern.compile("^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}Z$");
             List<String> listOfResults = ds
                     .select("new")
                     .collectAsList()
                     .stream()
                     .map(r -> r.getAs(0).toString())
                     .collect(Collectors.toList());
+            Assertions.assertEquals(12, listOfResults.size());
             for (String s : listOfResults) {
-                // match 00/00/0000 00:00:00
-                Matcher m = Pattern.compile("\\d{2}/\\d{2}/\\d{4} \\d{2}:\\d{2}:\\d{2}").matcher(s);
-
-                Assertions.assertTrue(m.find());
+                Matcher matcher = iso8601pattern.matcher(s);
+                Assertions.assertTrue(matcher.find());
             }
         });
     }
@@ -150,11 +150,13 @@ void testConvertCtime() {
                     .stream()
                     .map(r -> r.getAs(0).toString())
                     .collect(Collectors.toList());
+            Assertions.assertEquals(12, listOfResults.size());
             List<String> expectedResults = Arrays
                     .asList(
-                            "01/01/1970 00:00:11", "01/01/1970 00:00:11", "01/01/1970 00:00:10", "01/01/1970 00:00:09",
-                            "01/01/1970 00:00:08", "01/01/1970 00:00:07", "01/01/1970 00:00:06", "01/01/1970 00:00:05",
-                            "01/01/1970 00:00:04", "01/01/1970 00:00:03", "01/01/1970 00:00:02", "01/01/1970 00:00:01"
+                            "1970-01-01T00:00:11Z", "1970-01-01T00:00:11Z", "1970-01-01T00:00:10Z",
+                            "1970-01-01T00:00:09Z", "1970-01-01T00:00:08Z", "1970-01-01T00:00:07Z",
+                            "1970-01-01T00:00:06Z", "1970-01-01T00:00:05Z", "1970-01-01T00:00:04Z",
+                            "1970-01-01T00:00:03Z", "1970-01-01T00:00:02Z", "1970-01-01T00:00:01Z"
                     );
 
             for (int i = 0; i < listOfResults.size(); i++) {