self.

@card(id="eda", type="blank")
    @step
    def load_data(self):

        from datasets import load_dataset

        def walk_directory(root):
            path_keys = []
            for path, subdirs, files in os.walk(root):
                for name in files:
                # create a tuple of (key, path)
                    path_keys.append((os.path.relpath(os.path.join(path, name), root),os.path.join(path, name)))
            return path_keys

        logger.info(f'loading data')


        with S3() as s3:
            s3obj = s3.get(self.org_data)
            dataset = load_dataset("csv")
            print(dataset.shape)

        #saving dataset to s3
        os.mkdir('/flow/hfdataset')
        dataset.save_to_disk('/flow/hfdataset')
        with S3(s3root='<s3://bucket/hfdataset>') as s3:
            s3.put_files(walk_directory('/flow/hfdataset'))

        batch_size = 60
        num_batches = math.ceil(num_rows / batch_size)
        self.batches = [(i*batch_size, min((i+1)*batch_size, num_rows)) for i in range(num_batches)]

        self.next(self.hit_nightingale_model,foreach='batches')

    @step
    def hit_model(self):
        from datasets import load_from_disk
        import shutil

        def download_model(s3_path, download_path='/root/dir/'):
            final_path = s3_path
            os.makedirs('/root/dir', exist_ok=True)
            with S3(s3root=final_path) as s3:
                for s3obj in s3.get_all():
                    print('s3 key --->', s3obj.key)
                    move_path = os.path.join(download_path, s3obj.key)
                    print('move path --->', move_path)
                    if not os.path.exists(os.path.dirname(move_path)):
                        os.makedirs(os.path.dirname(move_path), exist_ok=True)
                    shutil.move(s3obj.path, os.path.join(download_path, s3obj.key))
                #print('dirpath ---->', (os.path.join(download_path, s3obj.key)))
                dataset=load_from_disk('/root/dir/')
                print(dataset)
                return dataset

        print('self.input --->', self.input)
        indices = self.input

        model = Model()
        dataset = download_model('<s3://bucket/hfdataset>')
        print(dataset)
        dataset = dataset['train'].select(range(indices[0],indices[1]))
        print(dataset)

        def process_row(row):
            result = model.run(report)
            return result

        self.result_dataset =  dataset.map(process_row, num_proc=32,batch_size=50000)
        self.next(self.join_results)

    @step
    def join_results(self,inputs):
        for  dataset in result_datasets:
            print(datset)
            print(dataset['train'][0])