feat(s3): add s3 source

metadata-ingestion/examples/recipes/s3_to_file.yml

@@ -0,0 +1,19 @@
+---
+# see https://datahubproject.io/docs/metadata-ingestion/source_docs/s3_data_lake for complete documentation
+source:
+  type: "s3"
+  config:
+    platform: s3
+    path_spec: 
+      include: "s3://covid19-lake/covid_knowledge_graph/csv/nodes/*.*"
+    aws_config:
+      aws_access_key_id: accessKey
+      aws_secret_access_key: secretKey
+      aws_region: us-east-2
+
+# see https://datahubproject.io/docs/metadata-ingestion/sink_docs/file for complete documentation
+sink:
+  type: "file"
+  config:
+    filename: "./s3_data_lake_mces.json"
+

metadata-ingestion/setup.py

@@ -113,14 +113,20 @@ def get_long_description():
     "tableschema>=1.20.2",
     "ujson>=4.3.0",
     "types-ujson>=4.2.1",
-    "smart-open[s3]>=5.2.1",
+    "smart-open[s3]>=5.2.1"
 }
 
 data_lake_profiling = {
     "pydeequ==1.0.1",
     "pyspark==3.0.3",
 }
 
+s3_base = {
+    *data_lake_base,
+    "moto[s3]",
+    "wcmatch",
+}
+
 # Note: for all of these, framework_common will be added.
 plugins: Dict[str, Set[str]] = {
     # Sink plugins.
@@ -142,6 +148,7 @@ def get_long_description():
     "datahub-lineage-file": set(),
     "datahub-business-glossary": set(),
     "data-lake": {*data_lake_base, *data_lake_profiling},
+    "s3": {*s3_base, *data_lake_profiling},
     "dbt": {"requests"},
     "druid": sql_common | {"pydruid>=0.6.2"},
     # Starting with 7.14.0 python client is checking if it is connected to elasticsearch client. If its not it throws
@@ -224,7 +231,7 @@ def get_long_description():
     *base_requirements,
     *framework_common,
     *mypy_stubs,
-    *data_lake_base,
+    *s3_base,
     "black>=21.12b0",
     "coverage>=5.1",
     "flake8>=3.8.3",
@@ -267,6 +274,7 @@ def get_long_description():
             "redshift",
             "redshift-usage",
             "data-lake",
+            "s3",
             "tableau",
             "trino",
             "hive",
@@ -335,6 +343,7 @@ def get_long_description():
         "clickhouse = datahub.ingestion.source.sql.clickhouse:ClickHouseSource",
         "clickhouse-usage = datahub.ingestion.source.usage.clickhouse_usage:ClickHouseUsageSource",
         "data-lake = datahub.ingestion.source.data_lake:DataLakeSource",
+        "s3 = datahub.ingestion.source.s3:S3Source",
         "dbt = datahub.ingestion.source.dbt:DBTSource",
         "druid = datahub.ingestion.source.sql.druid:DruidSource",
         "elasticsearch = datahub.ingestion.source.elastic_search:ElasticsearchSource",

metadata-ingestion/src/datahub/emitter/mcp_builder.py

@@ -57,6 +57,14 @@ class BigQueryDatasetKey(ProjectIdKey):
     dataset_id: str
 
 
+class FolderKey(PlatformKey):
+    folder_abs_path: str
+
+
+class S3BucketKey(PlatformKey):
+    bucket_name: str
+
+
 class DatahubKeyJSONEncoder(json.JSONEncoder):
 
     # overload method default
@@ -158,7 +166,7 @@ def gen_containers(
         # entityKeyAspect=ContainerKeyClass(guid=schema_container_key.guid()),
         aspectName="dataPlatformInstance",
         aspect=DataPlatformInstance(
-            platform=f"{make_data_platform_urn(container_key.platform)}"
+            platform=f"{make_data_platform_urn(container_key.platform)}",
         ),
     )
     wu = MetadataWorkUnit(

metadata-ingestion/src/datahub/ingestion/source/aws/s3_util.py

@@ -19,6 +19,10 @@ def strip_s3_prefix(s3_uri: str) -> str:
     )
 
 
+def get_bucket_relative_path(s3_uri: str) -> str:
+    return "/".join(strip_s3_prefix(s3_uri).split("/")[1:])
+
+
 def make_s3_urn(s3_uri: str, env: str) -> str:
 
     s3_name = strip_s3_prefix(s3_uri)
@@ -33,3 +37,11 @@ def make_s3_urn(s3_uri: str, env: str) -> str:
         return f"urn:li:dataset:(urn:li:dataPlatform:s3,{name}_{extension},{env})"
 
     return f"urn:li:dataset:(urn:li:dataPlatform:s3,{s3_name},{env})"
+
+
+def get_bucket_name(s3_uri: str) -> str:
+    if not is_s3_uri(s3_uri):
+        raise ValueError(
+            f"Not an S3 URI. Must start with one of the following prefixes: {str(S3_PREFIXES)}"
+        )
+    return strip_s3_prefix(s3_uri).split("/")[0]