Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(ingest): support for Avro data lake files #3913

Merged
merged 7 commits into from
Jan 26, 2022
Merged

feat(ingest): support for Avro data lake files #3913

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
d47e28f
Add avro support
kevinhu Jan 18, 2022
1972524
No ORC
kevinhu Jan 18, 2022
e2964ed
Black
kevinhu Jan 19, 2022
5141d64
Ignore failures
kevinhu Jan 19, 2022
1a92250
Flake8
kevinhu Jan 19, 2022
4dc6c3b
Merge branch 'master' of github.com:linkedin/datahub into avro-data-lake
kevinhu Jan 20, 2022
f2666b4
Merge branch 'master' of github.com:linkedin/datahub into avro-data-lake
kevinhu Jan 21, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions metadata-ingestion/source_docs/data_lake.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ This connector supports both local files as well as those stored on AWS S3 (whic
- TSV
- Parquet
- JSON
- Apache Avro

:::caution

Expand Down
27 changes: 22 additions & 5 deletions metadata-ingestion/src/datahub/ingestion/source/data_lake/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
StructType,
TimestampType,
)
from pyspark.sql.utils import AnalysisException

from datahub.emitter.mce_builder import make_data_platform_urn, make_dataset_urn
from datahub.emitter.mcp import MetadataChangeProposalWrapper
Expand Down Expand Up @@ -121,7 +122,16 @@ def __init__(self, config: DataLakeSourceConfig, ctx: PipelineContext):

conf = SparkConf()

conf.set("spark.jars.packages", "org.apache.hadoop:hadoop-aws:3.0.3")
conf.set(
"spark.jars.packages",
",".join(
[
"org.apache.hadoop:hadoop-aws:3.0.3",
"org.apache.spark:spark-avro_2.12:3.0.3",
pydeequ.deequ_maven_coord,
]
),
)

if self.source_config.aws_config is not None:

Expand Down Expand Up @@ -170,7 +180,6 @@ def __init__(self, config: DataLakeSourceConfig, ctx: PipelineContext):
"org.apache.hadoop.fs.s3a.AnonymousAWSCredentialsProvider",
)

conf.set("spark.jars.packages", pydeequ.deequ_maven_coord)
conf.set("spark.jars.excludes", pydeequ.f2j_maven_coord)
conf.set("spark.driver.memory", config.spark_driver_memory)

Expand Down Expand Up @@ -206,11 +215,19 @@ def read_file(self, file: str) -> Optional[DataFrame]:
)
elif file.endswith(".json"):
df = self.spark.read.json(file)
elif file.endswith(".avro"):
try:
df = self.spark.read.format("avro").load(file)
except AnalysisException:
self.report.report_warning(
file,
"To ingest avro files, please install the spark-avro package: https://mvnrepository.com/artifact/org.apache.spark/spark-avro_2.12/3.0.3",
)
return None

# TODO: add support for more file types
# elif file.endswith(".orc"):
# df = self.spark.read.orc(file)
# elif file.endswith(".avro"):
# df = self.spark.read.avro(file)
# df = self.spark.read.orc(file)
else:
self.report.report_warning(file, f"file {file} has unsupported extension")
return None
Expand Down
273 changes: 273 additions & 0 deletions metadata-ingestion/tests/integration/data_lake/data_lake_mces_golden.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1310,6 +1310,168 @@
"properties": null
}
},
{
"auditHeader": null,
"proposedSnapshot": {
"com.linkedin.pegasus2avro.metadata.snapshot.DatasetSnapshot": {
"urn": "urn:li:dataset:(urn:li:dataPlatform:data-lake-test,cases_1.chord_progressions_avro,PROD)",
"aspects": [
{
"com.linkedin.pegasus2avro.dataset.DatasetProperties": {
"customProperties": {},
"externalUrl": null,
"description": "",
"uri": null,
"tags": []
}
},
{
"com.linkedin.pegasus2avro.schema.SchemaMetadata": {
"schemaName": "chord_progressions_avro.avro",
"platform": "urn:li:dataPlatform:data-lake-test",
"version": 0,
"created": {
"time": 0,
"actor": "urn:li:corpuser:unknown",
"impersonator": null
},
"lastModified": {
"time": 0,
"actor": "urn:li:corpuser:unknown",
"impersonator": null
},
"deleted": null,
"dataset": null,
"cluster": null,
"hash": "",
"platformSchema": {
"com.linkedin.pegasus2avro.schema.OtherSchema": {
"rawSchema": ""
}
},
"fields": [
{
"fieldPath": "1st chord",
"jsonPath": null,
"nullable": false,
"description": null,
"type": {
"type": {
"com.linkedin.pegasus2avro.schema.NumberType": {}
}
},
"nativeDataType": "LongType",
"recursive": false,
"globalTags": null,
"glossaryTerms": null,
"isPartOfKey": false,
"jsonProps": null
},
{
"fieldPath": "2nd chord",
"jsonPath": null,
"nullable": false,
"description": null,
"type": {
"type": {
"com.linkedin.pegasus2avro.schema.NumberType": {}
}
},
"nativeDataType": "LongType",
"recursive": false,
"globalTags": null,
"glossaryTerms": null,
"isPartOfKey": false,
"jsonProps": null
},
{
"fieldPath": "3rd chord",
"jsonPath": null,
"nullable": false,
"description": null,
"type": {
"type": {
"com.linkedin.pegasus2avro.schema.NumberType": {}
}
},
"nativeDataType": "LongType",
"recursive": false,
"globalTags": null,
"glossaryTerms": null,
"isPartOfKey": false,
"jsonProps": null
},
{
"fieldPath": "4th chord",
"jsonPath": null,
"nullable": false,
"description": null,
"type": {
"type": {
"com.linkedin.pegasus2avro.schema.StringType": {}
}
},
"nativeDataType": "StringType",
"recursive": false,
"globalTags": null,
"glossaryTerms": null,
"isPartOfKey": false,
"jsonProps": null
},
{
"fieldPath": "Progression Quality",
"jsonPath": null,
"nullable": false,
"description": null,
"type": {
"type": {
"com.linkedin.pegasus2avro.schema.NumberType": {}
}
},
"nativeDataType": "DoubleType",
"recursive": false,
"globalTags": null,
"glossaryTerms": null,
"isPartOfKey": false,
"jsonProps": null
}
],
"primaryKeys": null,
"foreignKeysSpecs": null,
"foreignKeys": null
}
}
]
}
},
"proposedDelta": null,
"systemMetadata": {
"lastObserved": 1615443388097,
"runId": "data-lake-test",
"registryName": null,
"registryVersion": null,
"properties": null
}
},
{
"auditHeader": null,
"entityType": "dataset",
"entityUrn": "urn:li:dataset:(urn:li:dataPlatform:data-lake-test,cases_1.chord_progressions_avro,PROD)",
"entityKeyAspect": null,
"changeType": "UPSERT",
"aspectName": "datasetProfile",
"aspect": {
"value": "{\"timestampMillis\": 1615443388097, \"rowCount\": 28, \"columnCount\": 5, \"fieldProfiles\": [{\"fieldPath\": \"1st chord\", \"uniqueCount\": 5, \"uniqueProportion\": 0.17857142857142858, \"nullCount\": 0, \"nullProportion\": 0.0, \"distinctValueFrequencies\": [{\"value\": \"1\", \"frequency\": 5}, {\"value\": \"3\", \"frequency\": 1}, {\"value\": \"4\", \"frequency\": 10}, {\"value\": \"5\", \"frequency\": 7}, {\"value\": \"6\", \"frequency\": 5}], \"sampleValues\": [\"6\", \"4\", \"5\", \"1\", \"6\", \"4\", \"5\", \"4\", \"5\", \"4\", \"5\", \"1\", \"6\", \"1\", \"4\", \"3\", \"4\", \"4\", \"5\", \"1\"]}, {\"fieldPath\": \"2nd chord\", \"uniqueCount\": 7, \"uniqueProportion\": 0.25, \"nullCount\": 0, \"nullProportion\": 0.0, \"distinctValueFrequencies\": [{\"value\": \"1\", \"frequency\": 10}, {\"value\": \"2\", \"frequency\": 2}, {\"value\": \"3\", \"frequency\": 2}, {\"value\": \"4\", \"frequency\": 6}, {\"value\": \"5\", \"frequency\": 5}, {\"value\": \"6\", \"frequency\": 2}, {\"value\": \"7\", \"frequency\": 1}], \"sampleValues\": [\"4\", \"1\", \"6\", \"4\", \"2\", \"1\", \"1\", \"1\", \"4\", \"5\", \"5\", \"4\", \"2\", \"1\", \"1\", \"4\", \"3\", \"5\", \"4\", \"5\"]}, {\"fieldPath\": \"3rd chord\", \"uniqueCount\": 6, \"uniqueProportion\": 0.21428571428571427, \"nullCount\": 0, \"nullProportion\": 0.0, \"distinctValueFrequencies\": [{\"value\": \"1\", \"frequency\": 3}, {\"value\": \"3\", \"frequency\": 1}, {\"value\": \"4\", \"frequency\": 9}, {\"value\": \"5\", \"frequency\": 8}, {\"value\": \"6\", \"frequency\": 5}, {\"value\": \"7\", \"frequency\": 2}], \"sampleValues\": [\"5\", \"5\", \"5\", \"5\", \"4\", \"5\", \"4\", \"1\", \"4\", \"5\", \"1\", \"6\", \"5\", \"1\", \"4\", \"6\", \"6\", \"4\", \"3\", \"7\"]}, {\"fieldPath\": \"4th chord\", \"uniqueCount\": 20, \"uniqueProportion\": 0.7142857142857143, \"nullCount\": 0, \"nullProportion\": 0.0, \"distinctValueFrequencies\": [{\"value\": \"Alternative\", \"frequency\": 1}, {\"value\": \"Catchy\", \"frequency\": 1}, {\"value\": \"Didgeridoo\", \"frequency\": 1}, {\"value\": \"Dreadful\", \"frequency\": 2}, {\"value\": \"Endless\", \"frequency\": 1}, {\"value\": \"Energetic\", \"frequency\": 1}, {\"value\": \"Flamenco\", \"frequency\": 2}, {\"value\": \"Folk\", \"frequency\": 2}, {\"value\": \"Grunge\", \"frequency\": 1}, {\"value\": \"Jazz\", \"frequency\": 1}, {\"value\": \"Love\", \"frequency\": 1}, {\"value\": \"Memories\", \"frequency\": 1}, {\"value\": \"Moody\", \"frequency\": 2}, {\"value\": \"Pop\", \"frequency\": 2}, {\"value\": \"Rebellious\", \"frequency\": 1}, {\"value\": \"Sad\", \"frequency\": 3}, {\"value\": \"Simple\", \"frequency\": 2}, {\"value\": \"Sweet\", \"frequency\": 1}, {\"value\": \"Wildside\", \"frequency\": 1}, {\"value\": \"Wistful\", \"frequency\": 1}], \"sampleValues\": [\"Dreadful\", \"Memories\", \"Flamenco\", \"Rebellious\", \"Endless\", \"Alternative\", \"Folk\", \"Sweet\", \"Sad\", \"Sad\", \"Simple\", \"Wistful\", \"Dreadful\", \"Didgeridoo\", \"Simple\", \"Energetic\", \"Grunge\", \"Sad\", \"Flamenco\", \"Moody\"]}, {\"fieldPath\": \"Progression Quality\", \"uniqueCount\": 1, \"uniqueProportion\": 0.03571428571428571, \"nullCount\": 0, \"nullProportion\": 0.0, \"distinctValueFrequencies\": [{\"value\": \"NaN\", \"frequency\": 28}], \"sampleValues\": [\"nan\", \"nan\", \"nan\", \"nan\", \"nan\", \"nan\", \"nan\", \"nan\", \"nan\", \"nan\", \"nan\", \"nan\", \"nan\", \"nan\", \"nan\", \"nan\", \"nan\", \"nan\", \"nan\", \"nan\"]}]}",
"contentType": "application/json"
},
"systemMetadata": {
"lastObserved": 1615443388097,
"runId": "data-lake-test",
"registryName": null,
"registryVersion": null,
"properties": null
}
},
{
"auditHeader": null,
"proposedSnapshot": {
Expand Down Expand Up @@ -1566,6 +1728,117 @@
"properties": null
}
},
{
"auditHeader": null,
"proposedSnapshot": {
"com.linkedin.pegasus2avro.metadata.snapshot.DatasetSnapshot": {
"urn": "urn:li:dataset:(urn:li:dataPlatform:data-lake-test,cases_1.food_avro,PROD)",
"aspects": [
{
"com.linkedin.pegasus2avro.dataset.DatasetProperties": {
"customProperties": {},
"externalUrl": null,
"description": "",
"uri": null,
"tags": []
}
},
{
"com.linkedin.pegasus2avro.schema.SchemaMetadata": {
"schemaName": "food_avro.avro",
"platform": "urn:li:dataPlatform:data-lake-test",
"version": 0,
"created": {
"time": 0,
"actor": "urn:li:corpuser:unknown",
"impersonator": null
},
"lastModified": {
"time": 0,
"actor": "urn:li:corpuser:unknown",
"impersonator": null
},
"deleted": null,
"dataset": null,
"cluster": null,
"hash": "",
"platformSchema": {
"com.linkedin.pegasus2avro.schema.OtherSchema": {
"rawSchema": ""
}
},
"fields": [
{
"fieldPath": "weight",
"jsonPath": null,
"nullable": false,
"description": null,
"type": {
"type": {
"com.linkedin.pegasus2avro.schema.NumberType": {}
}
},
"nativeDataType": "LongType",
"recursive": false,
"globalTags": null,
"glossaryTerms": null,
"isPartOfKey": false,
"jsonProps": null
},
{
"fieldPath": "height",
"jsonPath": null,
"nullable": false,
"description": null,
"type": {
"type": {
"com.linkedin.pegasus2avro.schema.NumberType": {}
}
},
"nativeDataType": "LongType",
"recursive": false,
"globalTags": null,
"glossaryTerms": null,
"isPartOfKey": false,
"jsonProps": null
}
],
"primaryKeys": null,
"foreignKeysSpecs": null,
"foreignKeys": null
}
}
]
}
},
"proposedDelta": null,
"systemMetadata": {
"lastObserved": 1615443388097,
"runId": "data-lake-test",
"registryName": null,
"registryVersion": null,
"properties": null
}
},
{
"auditHeader": null,
"entityType": "dataset",
"entityUrn": "urn:li:dataset:(urn:li:dataPlatform:data-lake-test,cases_1.food_avro,PROD)",
"entityKeyAspect": null,
"changeType": "UPSERT",
"aspectName": "datasetProfile",
"aspect": {
"value": "{\"timestampMillis\": 1615443388097, \"rowCount\": 70, \"columnCount\": 2, \"fieldProfiles\": [{\"fieldPath\": \"weight\", \"uniqueCount\": 10, \"uniqueProportion\": 0.14285714285714285, \"nullCount\": 0, \"nullProportion\": 0.0, \"distinctValueFrequencies\": [{\"value\": \"10\", \"frequency\": 7}, {\"value\": \"2\", \"frequency\": 7}, {\"value\": \"23\", \"frequency\": 7}, {\"value\": \"32\", \"frequency\": 7}, {\"value\": \"36\", \"frequency\": 7}, {\"value\": \"43\", \"frequency\": 7}, {\"value\": \"49\", \"frequency\": 7}, {\"value\": \"50\", \"frequency\": 7}, {\"value\": \"53\", \"frequency\": 7}, {\"value\": \"72\", \"frequency\": 7}], \"sampleValues\": [\"23\", \"50\", \"10\", \"50\", \"49\", \"43\", \"10\", \"23\", \"72\", \"72\", \"23\", \"10\", \"72\", \"50\", \"32\", \"32\", \"72\", \"49\", \"36\", \"43\"]}, {\"fieldPath\": \"height\", \"uniqueCount\": 4, \"uniqueProportion\": 0.05714285714285714, \"nullCount\": 0, \"nullProportion\": 0.0, \"distinctValueFrequencies\": [{\"value\": \"4\", \"frequency\": 24}, {\"value\": \"5\", \"frequency\": 15}, {\"value\": \"6\", \"frequency\": 23}, {\"value\": \"7\", \"frequency\": 8}], \"sampleValues\": [\"4\", \"6\", \"6\", \"4\", \"5\", \"5\", \"5\", \"4\", \"4\", \"6\", \"6\", \"7\", \"7\", \"4\", \"6\", \"4\", \"5\", \"6\", \"5\", \"4\"]}]}",
"contentType": "application/json"
},
"systemMetadata": {
"lastObserved": 1615443388097,
"runId": "data-lake-test",
"registryName": null,
"registryVersion": null,
"properties": null
}
},
{
"auditHeader": null,
"proposedSnapshot": {
Expand Down
Binary file added ...data-ingestion/tests/integration/data_lake/test_data/cases_1/chord_progressions_avro.avro
View file
Open in desktop
Binary file not shown.
Binary file added metadata-ingestion/tests/integration/data_lake/test_data/cases_1/food_avro.avro
View file
Open in desktop
Binary file not shown.