Шаблон AWS Glue Grok, отметка времени в миллисекундах

Неправильное представление с классификаторами заключается в том, что они предназначены для указания форматов файлов в дополнение к встроенным, таким как JSON, CSV и т. д., а НЕ для указания форматов анализа отдельных типов данных.

Как предполагает пользователь @lilline, лучший способ изменить тип данных — использовать функцию ApplyMapping.

При создании задания Glue вы можете выбрать вариант: Предлагаемый скрипт, сгенерированный AWS Glue.

Затем, выбрав таблицу из каталога Glue в качестве источника, вы можете внести изменения в типы данных, имена столбцов и т. д.

Выходной код может выглядеть примерно так:

applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("paymentid", "string", "paymentid", "string"), ("updateddateutc", "string", "updateddateutc", "timestamp"), ...], transformation_ctx = "applymapping1")

Эффективное преобразование строки updateddateutc в отметку времени.

Чтобы создать классификатор, вам нужно будет указать каждый отдельный столбец в файле.

Classifier type: Grok 
Classification: Name Grok 
pattern: %{MY_TIMESTAMP} 
Custom patterns MY_TIMESTAMP (%{USERNAME:test}[,]%{YEAR:year}[-]%{MONTHNUM:mm}[-]%{MONTHDAY:dd} %{TIME:time})

Я также не смог понять, как это сделать с помощью классификаторов, но в итоге я преобразовал метку времени из строки в дату и время, написав пользовательское преобразование в скрипт сопоставления (python).

Ниже мой рабочий код. col2 — это столбец, указанный поисковым роботом в виде строки, и здесь я преобразовываю его в дату и время Python.

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

from datetime import datetime

args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "s3_events", table_name = "events", transformation_ctx = "datasource0")

def convert_dates(rec):
    rec["col2"] = datetime.strptime(rec["col2"], "%d.%m.%Y")
    return rec
custommapping1 = Map.apply(frame = datasource0, f = convert_dates, transformation_ctx = "custommapping1")

applymapping1 = ApplyMapping.apply(frame = custommapping1, mappings = [("col0", "string", "col0", "string"), ("col1", "string", "col1", "string"), ("col2", "date", "col2", "date")], transformation_ctx = "applymapping1")

selectfields2 = SelectFields.apply(frame = applymapping1, paths = ["col2", "col0", "col1"], transformation_ctx = "selectfields2")

resolvechoice3 = ResolveChoice.apply(frame = selectfields2, choice = "MATCH_CATALOG", database = "mydb", table_name = "mytable", transformation_ctx = "resolvechoice3")

resolvechoice4 = ResolveChoice.apply(frame = resolvechoice3, choice = "make_cols", transformation_ctx = "resolvechoice4")

datasink5 = glueContext.write_dynamic_frame.from_catalog(frame = resolvechoice4, database = "mydb", table_name = "mytable", transformation_ctx = "datasink5")
job.commit()