<jupyter_start><jupyter_text>Ingest Data From s3 to BTrDBThis notebook will demonstrate how to subclass `CloudMixin` to create a `DataParser` that can load data from AWS s3.<jupyter_code>import os
import s3fs
import btrdb
import pandas as pd

from pgimport.csv_parser import MyCSVParser
from pgimport.ingest import DataIngestor
from pgimport.cloud import S3Mixin, S3File<jupyter_output><empty_output><jupyter_text>Creating a Custom FileWe can extend the `File` object to include a `count`, which can be passed to a `DataIngestor` and allows for a progress bar during ingestion.<jupyter_code>class S3ProgBarFile(S3File):
    def __init__(self, path, count=True, header=False):
        super().__init__(path)
        # TODO: this works but its slow, need to find a more efficient way to do this
        if count:
            header_rows = 1 if header else 0
            df = pd.read_csv(path)
            self.count = (len(df) - header_rows) * (len(df.columns) - 1)<jupyter_output><empty_output><jupyter_text>Creating a Custom DataParserWe can create a custom `DataParser` by extending the `MyCSVParser` and combining it with the `S3Mixin`, which provides a connection to s3 by assigning a `client` attribute that is equivalent to `boto3.client("s3")`<jupyter_code>class S3CSVParser(S3Mixin, MyCSVParser):
    """
    Parameters
    -----------
    collection_prefix: str
        prefix to add to all streams' collection names
    regex: str
        regex string to use to retrieve collection from file name
    metadata: dict or str
        either a dict of metadata or a str filename referring to a yaml/json metadata file
    meta_func:
        callback function to use to map metadata to Stream objects
    bucket: str
        s3 bucket that contains raw data files
    s3_prefix: str
        (optional) subdirectory within bucket that contains raw data files 
    **kwargs: dict
        (optional) key/value pairs specifying required AWS credentials. Can be left as None
        if credentials are stored as environtment variables
    """
    def __init__(self, collection_prefix=None, regex=None, metadata=None, meta_func=None, bucket=None, s3_prefix=None, **kwargs):
        MyCSVParser.__init__(self, None, collection_prefix=collection_prefix, regex=regex, metadata=metadata, meta_func=meta_func)
        # establish attrs that are specific to this class
        self.bucket = bucket
        self.s3_prefix = s3_prefix
        # init S3Mixin, connects to s3 and provides a handle via self.client
        S3Mixin.__init__(self, **kwargs)
        self.client = self.connect()
    
    def collect_files(self):
        return [S3ProgBarFile(fpath, header=True) for fpath in self.list_objects(self.bucket, prefix=self.s3_prefix)]<jupyter_output><empty_output><jupyter_text>Ingest s3 Data Next we will use the `S3CSVParser` to load data from s3, create `Stream` objects, and pass them to the `DataIngestor` for insertion into BTrDB.<jupyter_code>bucket = "your-bucket"
subdir = "subdirectory-within-bucket"

# explicity setting credentials is optional, you can also just store them as env vars
aws_creds = {
    "aws_access_key_id": os.environ["AWS_ACCESS_KEY_ID"],
    "aws_secret_access_key": os.environ["AWS_SECRET_ACCESS_KEY"],
    "aws_session_token": os.environ["AWS_SESSION_TOKEN"]
}

s3 = S3CSVParser(collection_prefix="test_ingest", bucket=bucket, s3_prefix=subdir, **aws_creds)
files = s3.collect_files()
total_points = sum([f.count for f in files])

print(f"Found {len(files)} files")
print(f"Files contain {total_points} points")
# replace with your BTrDB credentials
conn = btrdb.connect(os.environ["BTRDB_ENDPOINTS"], apikey=os.environ["BTRDB_API_KEY"])

ingestor = DataIngestor(conn, total_points=total_points)
for stream in s3.instantiate_streams(files):
    ingestor.ingest(stream)<jupyter_output>6480060it [05:20, 27569.91it/s]