问读取数据时出错，错误消息: JSON表遇到太多错误，放弃。行

EN

RuntimeError: BigQuery job beam_bq_job_LOAD_AUTOMATIC_JOB_NAME_LOAD_STEP_614_c4a563c648634e9dbbf7be3a56578b6d_2f196decc8984a0d83dee92e19054ffb failed. Error Result: <ErrorProto
 location: 'gs://dataflow4bigquery/temp/bq_load/06bfafaa9dbb47338ad4f3a9914279fe/dotted-transit-351803.test_dataflow.inner_join/f714c1ac-c234-4a37-bf51-c725a969347a'
 message: 'Error while reading data, error message: JSON table encountered too many errors, giving up. Rows: 1; errors: 1. Please look into the errors[] collection for more details.'
 reason: 'invalid'> [while running 'WriteToBigQuery/BigQueryBatchFileLoads/WaitForDestinationLoadJobs']

from apache_beam.io.gcp.internal.clients import bigquery
import apache_beam as beam

def retTuple(element):
  
  thisTuple=element.split(',')
  return (thisTuple[0],thisTuple[1:])

def jstr(cstr):
    import datetime

    
    left_dict=cstr[1]['dep_data']
    right_dict=cstr[1]['loc_data']
    for i  in left_dict:
    
        for j in right_dict:
            id,name,rank,dept,dob,loc,city=([cstr[0]]+i+j)
            
            json_str={ "id":id,"name":name,"rank":rank,"dept":dept,"dob":datetime.datetime.strptime(dob, "%d-%m-%Y").strftime("%Y-%m-%d").strip("'"),"loc":loc,"city":city }
    return json_str
            
table_spec = 'dotted-transit-351803:test_dataflow.inner_join'
table_schema = 'id:INTEGER,name:STRING,rank:INTEGER,dept:STRING,dob:STRING,loc:INTEGER,city:STRING'   
gcs='gs://dataflow4bigquery/temp/'
    
p1 = beam.Pipeline()

# Apply a ParDo to the PCollection "words" to compute lengths for each word.
dep_rows = ( 
                p1
                | "Reading File 1" >> beam.io.ReadFromText('dept_data.txt')
                | 'Pair each employee with key' >> beam.Map(retTuple)          # {149633CM : [Marco,10,Accounts,1-01-2019]}
    
               )


loc_rows = ( 
                p1
                | "Reading File 2" >> beam.io.ReadFromText('location.txt') 
                | 'Pair each loc with key' >> beam.Map(retTuple)                # {149633CM : [9876843261,New York]}
               )


results = ({'dep_data': dep_rows, 'loc_data': loc_rows} 
           
           | beam.CoGroupByKey()
           | beam.Map(jstr)
           |  beam.io.WriteToBigQuery(
               custom_gcs_temp_location=gcs,
            table=table_spec,
            schema=table_schema,
            write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE,
            create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
            additional_bq_parameters={'timePartitioning': {'type': 'DAY'}}
               
        )
          )




 
p1.run().wait_until_finish()