Enabling readahead cache for s3fs.

Author: igorborgest

awswrangler/_utils.py

@@ -136,9 +136,9 @@ def get_fs(
     fs: s3fs.S3FileSystem = s3fs.S3FileSystem(
         anon=False,
         use_ssl=True,
-        default_cache_type="none",
+        default_cache_type="readahead",
         default_fill_cache=False,
-        default_block_size=134_217_728,  # 128 MB (50 * 2**20)
+        default_block_size=1_073_741_824,  # 1024 MB (1024 * 2**20)
         config_kwargs={"retries": {"max_attempts": 15}},
         session=ensure_session(session=session)._session,  # pylint: disable=protected-access
         s3_additional_kwargs=s3_additional_kwargs,

awswrangler/s3.py

@@ -1693,6 +1693,7 @@ def read_parquet(
         boto3_session=boto3_session,
         s3_additional_kwargs=s3_additional_kwargs,
     )
+    _logger.debug("pyarrow.parquet.ParquetDataset initialized.")
     if chunked is False:
         return _read_parquet(
             data=data, columns=columns, categories=categories, use_threads=use_threads, validate_schema=validate_schema
@@ -1710,13 +1711,17 @@ def _read_parquet(
     validate_schema: bool = True,
 ) -> pd.DataFrame:
     tables: List[pa.Table] = []
+    _logger.debug("Reading pieces...")
     for piece in data.pieces:
         table: pa.Table = piece.read(
             columns=columns, use_threads=use_threads, partitions=data.partitions, use_pandas_metadata=False
         )
+        _logger.debug("Appending piece in the list...")
         tables.append(table)
     promote: bool = not validate_schema
+    _logger.debug("Concating pieces...")
     table = pa.lib.concat_tables(tables, promote=promote)
+    _logger.debug("Converting PyArrow table to Pandas DataFrame...")
     return table.to_pandas(
         use_threads=use_threads,
         split_blocks=True,

testing/test_awswrangler/test_data_lake.py

@@ -1300,18 +1300,18 @@ def test_catalog_versioning(bucket, database):
 
     # Version 1
     df = pd.DataFrame({"c1": ["foo", "boo"]})
-    paths = wr.s3.to_parquet(
+    paths1 = wr.s3.to_parquet(
         df=df, path=path, dataset=True, database=database, table=table, mode="overwrite", catalog_versioning=True
     )["paths"]
-    wr.s3.wait_objects_exist(paths=paths, use_threads=False)
+    wr.s3.wait_objects_exist(paths=paths1, use_threads=False)
     df = wr.athena.read_sql_table(table=table, database=database)
     assert len(df.index) == 2
     assert len(df.columns) == 1
     assert str(df.c1.dtype) == "string"
 
     # Version 2
     df = pd.DataFrame({"c1": [1.0, 2.0]})
-    paths = wr.s3.to_csv(
+    paths2 = wr.s3.to_csv(
         df=df,
         path=path,
         dataset=True,
@@ -1321,15 +1321,16 @@ def test_catalog_versioning(bucket, database):
         catalog_versioning=True,
         index=False,
     )["paths"]
-    wr.s3.wait_objects_exist(paths=paths, use_threads=False)
+    wr.s3.wait_objects_exist(paths=paths2, use_threads=False)
+    wr.s3.wait_objects_not_exist(paths=paths1, use_threads=False)
     df = wr.athena.read_sql_table(table=table, database=database)
     assert len(df.index) == 2
     assert len(df.columns) == 1
     assert str(df.c1.dtype).startswith("float")
 
     # Version 3 (removing version 2)
     df = pd.DataFrame({"c1": [True, False]})
-    paths = wr.s3.to_csv(
+    paths3 = wr.s3.to_csv(
         df=df,
         path=path,
         dataset=True,
@@ -1339,7 +1340,8 @@ def test_catalog_versioning(bucket, database):
         catalog_versioning=False,
         index=False,
     )["paths"]
-    wr.s3.wait_objects_exist(paths=paths, use_threads=False)
+    wr.s3.wait_objects_exist(paths=paths3, use_threads=False)
+    wr.s3.wait_objects_not_exist(paths=paths2, use_threads=False)
     df = wr.athena.read_sql_table(table=table, database=database)
     assert len(df.index) == 2
     assert len(df.columns) == 1