original_docstring: 
          Read data from S3 datastore and process/transform/wrangle
          
docstring: Read data from S3 datastore and process/transform/wrangle
docstring_tokens: Read data from S3 datastore and process / transform / wrangle
code: def process_raw_data(self):
          """
          Read data from S3 datastore and process/transform/wrangle
          """
          import os
          from process_raw_data import process_raw_data
          from utils import get_filename
          from metaflow.metaflow_config import DATATOOLS_S3ROOT
  
          # get dataset S3 paths
          DATASET_PATH = os.path.join(
              DATATOOLS_S3ROOT, os.getenv('PARQUET_S3_PATH'))
          SEARCH_TRAIN_PATH = os.path.join(DATASET_PATH, get_filename(
              os.getenv('SEARCH_TRAIN_PATH')) + '.parquet')
          BROWSING_TRAIN_PATH = os.path.join(DATASET_PATH, get_filename(
              os.getenv('BROWSING_TRAIN_PATH')) + '.parquet')
          SKU_TO_CONTENT_PATH = os.path.join(DATASET_PATH, get_filename(
              os.getenv('SKU_TO_CONTENT_PATH')) + '.parquet')
  
          # process raw data
          processed_data = process_raw_data(search_train_path=SEARCH_TRAIN_PATH,
                                            browsing_train_path=BROWSING_TRAIN_PATH,
                                            sku_to_content_path=SKU_TO_CONTENT_PATH)
  
          # save as parquet onto S3
          with S3(run=self) as s3:
              # save s3 paths in dict
              self.data_paths = dict()
              s3_root = s3._s3root
              for name, data in processed_data.items():
                  data_path = os.path.join(s3_root, name + '.parquet')
                  data.to_parquet(path=data_path, engine='pyarrow')
                  self.data_paths[name] = data_path
  
          print(self.data_paths)
  
          self.next(self.data_validation)
code_tokens: def process_raw_data ( self ) : import os from process_raw_data import process_raw_data from utils import get_filename from metaflow . metaflow_config import DATATOOLS_S3ROOT DATASET_PATH = os . path . join ( DATATOOLS_S3ROOT , os . getenv ( 'PARQUET_S3_PATH' ) ) SEARCH_TRAIN_PATH = os . path . join ( DATASET_PATH , get_filename ( os . getenv ( 'SEARCH_TRAIN_PATH' ) ) + '.parquet' ) BROWSING_TRAIN_PATH = os . path . join ( DATASET_PATH , get_filename ( os . getenv ( 'BROWSING_TRAIN_PATH' ) ) + '.parquet' ) SKU_TO_CONTENT_PATH = os . path . join ( DATASET_PATH , get_filename ( os . getenv ( 'SKU_TO_CONTENT_PATH' ) ) + '.parquet' ) processed_data = process_raw_data ( search_train_path = SEARCH_TRAIN_PATH , browsing_train_path = BROWSING_TRAIN_PATH , sku_to_content_path = SKU_TO_CONTENT_PATH ) with S3 ( run = self ) as s3 : self . data_paths = dict ( ) s3_root = s3 . _s3root for name , data in processed_data . items ( ) : data_path = os . path . join ( s3_root , name + '.parquet' ) data . to_parquet ( path = data_path , engine = 'pyarrow' ) self . data_paths [ name ] = data_path print ( self . data_paths ) self . next ( self . data_validation )
short_docstring: Read data from S3 datastore and process/transform/wrangle
short_docstring_tokens: Read data from S3 datastore and process / transform / wrangle
comment: """
          Read data from S3 datastore and process/transform/wrangle
          """$SEP$# get dataset S3 paths$SEP$# process raw data$SEP$# save as parquet onto S3$SEP$# save s3 paths in dict