PipelineSteps.kt

fun toTableRow(record: Record) = TableRow().also {
    it.set("station", record.station)
    it.set("timestamp", Utils.toTimeStampBQFormat(record.timestamp))
    it.set("pm25", record.pm25)
    it.set("pm10", record.pm10)
    it.set("so2", record.so2)
    it.set("no2", record.no2)
    it.set("co", record.co)
    it.set("o3", record.o3)
    it.set("temp", record.temp)
    it.set("press", record.press)
    it.set("dewp", record.dewp)
    it.set("rain", record.rain)
    it.set("wd", record.wd)
    it.set("wspm", record.wspm)
}
fun mapEntityToTableRow(): MapElements<Record, TableRow> {
    return MapElements.into(TypeDescriptor.of(TableRow::class.java))
            .via(ProcessFunction { RecordSchema.toTableRow(it) })
}