GCP Dataflow Pipeline'ında Veri Dönüşüm Hataları Çözme

1import apache_beam as beam
2from apache_beam.options.pipeline_options import PipelineOptions
3import json
4import logging
5
6class TransformData(beam.DoFn):
7def process(self, element):
8try:
9parts = element.split(',')
10if len(parts)!= 3:
11raise ValueError(f"Beklenen 3 alan, {len(parts)} bulundu")
12
13user_id, name, age = parts
14age = int(age)
15
16if age < 0 or age > 150:
17raise ValueError(f"Geçersiz yaş: {age}")
18
19yield {
20'user_id': user_id.strip(),
21'name': name.strip(),
22'age': age
23}
24except Exception as e:
25logging.error(f"Dönüşüm hatası - {element}: {str(e)}")
26yield beam.pvalue.TaggedOutput('dead_letter', {
27'raw_data': element,
28'error': str(e)
29})
30
31def run():
32options = PipelineOptions()
33
34with beam.Pipeline(options=options) as p:
35lines = p | 'Oku' >> beam.io.ReadFromText('gs://bucket/input.csv')
36
37results = (lines 
38| 'Dönüştür' >> beam.ParDo(TransformData()).with_outputs(
39'dead_letter', main='main'))
40
41(results['main'] 
42| 'JSON Yaz' >> beam.Map(lambda x: json.dumps(x))
43| 'Başarılı Yaz' >> beam.io.WriteToText('gs://bucket/output.json'))
44
45(results['dead_letter']
46| 'Hata Yaz' >> beam.Map(lambda x: json.dumps(x))
47| 'Hata Dosyası' >> beam.io.WriteToText('gs://bucket/errors.json'))
48
49if __name__ == '__main__':
50run()

1python pipeline.py --runner DataflowRunner --project PROJECT_ID --temp_location gs://bucket/temp