AWS Glue job bookmark создает дубликаты для CSV-файлов - PullRequest
Новая
       80

AWS Glue job bookmark создает дубликаты для CSV-файлов

0 голосов
/ 24 июня 2019

Мы получаем 1 CSV-файл каждый день в S3 ведро от нашего поставщика в 11 утра. Я конвертирую этот файл в формат паркета, используя клей в 11:30 утра.

Я включил закладку задания, чтобы не обрабатывать уже обработанные файлы. Тем не менее, я вижу, что некоторые файлы перерабатываются, создавая дубликаты.

Я прочитал эти вопросы и ответы AWS Glue Bookmark создает дубликаты для PARQUET и AWS Glue Job Bookmarking объяснение

Они дали хорошее представление о закладках работы, но все еще не решают проблему.

Документация AWS гласит, что он поддерживает файлы CSV для закладок Документация AWS .

Хотите знать, если кто-нибудь поможет мне понять, в чем может быть проблема, и, если возможно, решение?)

Edit:

Вставка примера кода здесь в соответствии с просьбой Прабхакара. 

staging_database_name = "my-glue-db"
s3_target_path = "s3://mybucket/mydata/"


"""
 'date_index': date location in the file name
 'date_only': only date column is inserted
 'date_format': format of date
 'path': sub folder name in master bucket
"""

#fouo classified files
tables_spec = {
'sample_table': {'path': 'sample_table/load_date=','pkey': 'mykey', 'orderkey':'myorderkey'}
}

spark_conf = SparkConf().setAll([
  ("spark.hadoop.fs.s3.enableServerSideEncryption", "true"),
  ("spark.hadoop.fs.s3.serverSideEncryption.kms.keyId", kms_key_id)
])
sc = SparkContext(conf=spark_conf)

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

for table_name, spec in tables_spec.items():
    datasource0 = glueContext.create_dynamic_frame.from_catalog(database=database_name,
                                                                table_name=table_name,
                                                                transformation_ctx='datasource0')

    resolvechoice2 = ResolveChoice.apply(frame=datasource0, choice="make_struct", transformation_ctx='resolvechoice2')

    # Create spark data frame with input_file_name column
    delta_df = resolvechoice2.toDF().withColumn('ingest_datetime', lit(str(ingest_datetime)))

    date_dyf = DynamicFrame.fromDF(delta_df, glueContext, "date_dyf")
    master_folder_path1 = os.path.join(s3_target_path, spec['path']).replace('\\', '/')

    master_folder_path=master_folder_path1+load_date
    datasink4 = glueContext.write_dynamic_frame.from_options(frame=date_dyf,
                                                            connection_type='s3',
                                                            connection_options={"path": master_folder_path},
                                                            format='parquet', transformation_ctx='datasink4')
job.commit()
...