count = 0
ln = len(bkg_lst)
start = 0
for i in bkg_lst[:]: # list contains paths to files
part_bkg = sqlContext.read.option("delimiter", ";").format('com.databricks.spark.csv').options(header='false', inferschema='true').load(i)
if len(part_bkg.head(1)) != 0:
# filter out unnessesary columns
part_bkg = part_bkg.select('BKG_ID', 'PNR_ID', 'PAX_ID', 'SEG_NBR', 'PNR_CRTD_DT', 'OPRTNG_FLT_DT_UTC', 'TRAVELER_ORIG_TERMINAL', 'TRAVELER_DEST_TERMINAL', 'DEL_FLG', 'SRC_SYS_CRTD_TMSTMP')
part_bkg.createOrReplaceTempView("bkg")
# here goes nessesary SQL select, not heavy for sure
res_df = sqlContext.sql("""select b.ucp_id from ...""")
if len(res_df.head(1)) != 0:
res_df.select('UCP_ID', 'TRAVELER_DEST_TERMINAL').intersect(df_rec.select('UCP_ID','ITEM_KEY')).select('UCP_ID').coalesce(1).write.save(path='/tmp/true_recs/' +str(count) , format='csv', mode='append', sep=',')
count +=1
print(count, " of ", ln)<br>
... View more
