1  import sys
   2  from awsglue.transforms import *
   3  from awsglue.utils import getResolvedOptions
   4  from pyspark.context import SparkContext
   5  from awsglue.context import GlueContext
   6  from awsglue.job import Job
   7  from awsglue.dynamicframe import DynamicFrame
   8  import time
   9  from datetime import datetime, timedelta
  10  import requests
  11  import re
  12  from pyspark.sql.functions import *
  13  import boto3
  14  import datetime
  15  from credentials import CredentialManager
  16  from dataextractor import DataExtractor
  17  
  18  args = getResolvedOptions(sys.argv, ['JOB_NAME'])
  19  glueContext = GlueContext(SparkContext.getOrCreate())
  20  spark = glueContext.spark_session
  21  job = Job(glueContext)
  22  job.init(args['JOB_NAME'], args)
  23  
  24  
  25  spark.conf.set("spark.sql.optimizer.pushedFilters", "true")
  26  
  27  def snow():
  28      options={'table':'CUSTOMER_CATALOG',
  29               'warehouse':'COMPUTE', 'schema':'BUNGEE', 'database':'PETCO_BUNGEE_SHARE_INT'
  30               }
  31      source ="petco-snowflake"
  32      cred_manager = CredentialManager()
  33      pushdown_predicate=  "cast(product_id as varchar) in ('2778692', '2254733', '1354906', '3186808')"
  34      data_extractor = DataExtractor( glueContext,cred_manager)
  35      df = data_extractor.extract_data(source, options, pushdown_predicate, columns='product_id')
  36      #{'user': 'us_east_1', 'password': 'ntc*>*tS>3,",HM"', 'account': 'gztlgpj-us_east_1'}
  37      
  38      #print('petco count', df.count())
  39      return df
  40  
  41  
  42  def aurora():
  43      source ="retailscape-aurora"
  44      options={'table':'product_gaps_chewy_20230102', 'database':'assortment_analytics'}
  45      
  46      cred_manager = CredentialManager()
  47      pushdown_predicate=  "cast(sku as varchar) in ('3338349', '3079172', '967750', '159352')"
  48      
  49      data_extractor = DataExtractor( glueContext,cred_manager)
  50      df = data_extractor.extract_data(source, options, pushdown_predicate, columns ='bungee_product_id, source, sku, merch1')
  51      return df
  52  
  53  def glue():
  54      source ="glue"
  55      options={'table':'final_ingestion', 'database':'assortment'}
  56      
  57      cred_manager = CredentialManager()
  58      pushdown_predicate=  "load_Date = cast('2023-01-04' as date ) and cast(sku as varchar) in ('3338349', '3079172', '967750', '159352')"
  59      
  60      data_extractor = DataExtractor( glueContext,cred_manager)
  61      df = data_extractor.extract_data(source, options, pushdown_predicate, columns ='bungee_product_id, source, sku, merch1')
  62      return df
  63  
  64  
  65  
  66  
  67  s= datetime.datetime.now()
  68  df= snow()
  69  print(datetime.datetime.now()-s)
  70  df.show()
  71  print(df.select('product_id').collect())
  72  
  73  s= datetime.datetime.now()
  74  df= aurora()
  75  print(datetime.datetime.now()-s)
  76  df.show()
  77  print(df.select('bungee_product_id', 'sku').collect())
  78  
  79  s= datetime.datetime.now()
  80  df= glue()
  81  print(datetime.datetime.now()-s)
  82  df.show()
  83  print(df.select('source', 'sku').collect())
  84  
  85  
  86  
  87  
  88  job.commit()