feat(ingestion): schema inference for jsonlines in S3

metadata-ingestion/src/datahub/ingestion/source/schema_inference/json.py

@@ -1,5 +1,7 @@
+import logging
 from typing import IO, Dict, List, Type, Union
 
+import jsonlines as jsl
 import ujson
 
 from datahub.ingestion.source.schema_inference.base import SchemaInferenceBase
@@ -27,11 +29,18 @@
     "mixed": UnionTypeClass,
 }
 
+logger = logging.getLogger(__name__)
+
 
 class JsonInferrer(SchemaInferenceBase):
     def infer_schema(self, file: IO[bytes]) -> List[SchemaField]:
-
-        datastore = ujson.load(file)
+        try:
+            datastore = ujson.load(file)
+        except ujson.JSONDecodeError as e:
+            logger.info(f"Got ValueError: {e}. Retry with jsonlines")
+            file.seek(0)
+            reader = jsl.Reader(file)
+            datastore = [obj for obj in reader.iter(type=dict, skip_invalid=True)]
 
         if not isinstance(datastore, list):
             datastore = [datastore]