feat(spark-lineage): add support to custom env and platform_instance #4208

MugdhaHardikar-GSLab · 2022-02-22T09:35:59Z

Checklist

The PR conforms to DataHub's Contributing Guideline (particularly Commit Message Format)
Links to related issues (if applicable)
Tests for the changes have been added/updated (if applicable)
Docs related to the changes have been added/updated (if applicable)

…ab/datahub into spark-logging

refactor(spark-lineage): enhance logging and documentation

github-actions · 2022-02-22T09:59:32Z

Unit Test Results (build & test)

  75 files +  5   75 suites +5 15m 33s ⏱️ + 1m 52s
623 tests +12 563 ✔️ +11 59 💤 ±0 1 ❌ +1

For more details on these failures, see this check.

Results for commit c82e4fb. ± Comparison against base commit c6cb549.

♻️ This comment has been updated with latest results.

shirshanka

Partial review

shirshanka · 2022-02-23T04:37:59Z

metadata-integration/java/spark-lineage/src/main/java/datahub/spark/DatasetExtractor.java

@@ -14,6 +15,7 @@

 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
+import org.apache.hive.com.esotericsoftware.minlog.Log;


seems like a bad import

shirshanka · 2022-02-23T04:42:33Z

metadata-integration/java/spark-lineage/src/main/java/datahub/spark/DatasetExtractor.java

+    try {
+      fabricType = FabricType.valueOf(fabricTypeString);
+    } catch (IllegalArgumentException e) {
+      Log.warn("Invalid env ({}). Setting env to default PROD", DATASET_ENV_KEY);


parameter here should be fabricTypeString not DATASET_ENV_KEY

shirshanka · 2022-02-23T04:43:24Z

metadata-integration/java/spark-lineage/src/main/java/datahub/spark/DatasetExtractor.java

+      InsertIntoHadoopFsRelationCommand.class, SaveIntoDataSourceCommand.class,
+      CreateDataSourceTableAsSelectCommand.class, CreateHiveTableAsSelectCommand.class, InsertIntoHiveTable.class);
+  private static final String DATASET_ENV_KEY = "metadata.dataset.env";
+  private static final String DATSET_PLATFORM_INSTANCE_KEY = "metadata.dataset.platform_instance";


typo in constant name DATSET_ versus DATASET_

shirshanka · 2022-02-23T04:44:18Z

metadata-integration/java/spark-lineage/src/main/java/datahub/spark/DatasetExtractor.java

+      InsertIntoHadoopFsRelationCommand.class, SaveIntoDataSourceCommand.class,
+      CreateDataSourceTableAsSelectCommand.class, CreateHiveTableAsSelectCommand.class, InsertIntoHiveTable.class);
+  private static final String DATASET_ENV_KEY = "metadata.dataset.env";
+  private static final String DATSET_PLATFORM_INSTANCE_KEY = "metadata.dataset.platform_instance";


suggest you use camel case here to align with Java / Scala style..
so metadata.dataset.platformInstance as the config key.

shirshanka · 2022-02-23T04:47:17Z

metadata-integration/java/spark-lineage/src/main/java/datahub/spark/DatasetExtractor.java

      CreateDataSourceTableAsSelectCommand cmd = (CreateDataSourceTableAsSelectCommand) p;
      // TODO what of cmd.mode()
-      return Optional.of(new CatalogTableDataset(cmd.table()));
+      return Optional.of(new CatalogTableDataset(cmd.table(), getCommonPlatformInstance(datahubConfig), getCommonFabricType(datahubConfig)));


is there a good reason to compute the fabricType and platformInstance everytime, versus just calculating it once and using a final variable in the other method calls?

We need to extend this further to each dataset level platforminstance. So there will be current dataset specific selection. Thats why i have not get it done at common place

This would probably look a bit more clean with a true Visitor pattern rather than doing it through a map. Having this huge static initializer with a bunch of lambdas is pretty hard to read and is more error prone to missing an update to one of the lambdas.

metadata-integration/java/spark-lineage/src/main/java/datahub/spark/model/AppStartEvent.java

shirshanka · 2022-02-25T00:08:10Z

...ta-integration/java/spark-lineage/src/main/java/datahub/spark/model/dataset/JdbcDataset.java

+  //TODO: Should map to the central location on datahub for platform names
+  private static final Map<String, String> PLATFORM_NAME_MAPPING = new HashMap<>();
+  static {
+    PLATFORM_NAME_MAPPING.put("postgresql", "postgres");


why not add more mappings here?

For others url to platform name is same hence have not added them there.

RyanHolstien

Some clean-up comments, but overall LGTM. I don't think the visitor pattern one is a blocker, but if it's quick to do would make it more readable.

RyanHolstien · 2022-03-11T17:25:58Z

metadata-integration/java/spark-lineage/README.md

+
+| Field                                           | Required | Default | Description                                                             |
+|-------------------------------------------------|----------|---------|-------------------------------------------------------------------------|
+| spark.jars.packages                              | ✅        |         | Set with latest/required version  io.acryl:datahub-spark-lineage:0.8.23 |


Is this version still correct? Might be better to just leave it off if not

Version is not latest, but we dont update documentation on every release and put a note to check for latest version

Right, but it's better to just not specify a specific version then

RyanHolstien · 2022-03-11T18:49:00Z

metadata-integration/java/spark-lineage/src/main/java/datahub/spark/DatasetExtractor.java

      CreateDataSourceTableAsSelectCommand cmd = (CreateDataSourceTableAsSelectCommand) p;
      // TODO what of cmd.mode()
-      return Optional.of(new CatalogTableDataset(cmd.table()));
+      return Optional.of(new CatalogTableDataset(cmd.table(), getCommonPlatformInstance(datahubConfig), getCommonFabricType(datahubConfig)));


This would probably look a bit more clean with a true Visitor pattern rather than doing it through a map. Having this huge static initializer with a bunch of lambdas is pretty hard to read and is more error prone to missing an update to one of the lambdas.

RyanHolstien · 2022-03-11T18:53:06Z

metadata-integration/java/spark-lineage/src/main/java/datahub/spark/model/AppStartEvent.java


+    if (this.pipelineConfig.hasPath(PLATFORM_INSTANCE_KEY)) {
+      try {
+        DataPlatformInstance dpi = new DataPlatformInstance().setPlatform(new DataPlatformUrn("spark")).setInstance(


Pull repeated strings into constants

Using static inits vs visitor pattern was causios decision. Will discuss and take it up in another PR

RyanHolstien · 2022-03-11T18:59:12Z

metadata-integration/java/spark-lineage/src/main/java/datahub/spark/model/LineageUtils.java

@@ -25,6 +38,10 @@ private LineageUtils() {

  }

+  public static Urn dataPlatformInstanceUrn(String platform, String instance) throws URISyntaxException {
+    return new Urn("urn:li:dataPlatformInstance:(" + new DataPlatformUrn(platform).toString() + "," + instance + ")");


This can be done using the entityType & TupleKey constructor to avoid string manipulation and increased overhead from the Urn(String) constructor.

RyanHolstien · 2022-03-11T19:01:42Z

metadata-integration/java/spark-lineage/src/main/java/datahub/spark/model/LineageUtils.java

+  public static Urn dataPlatformInstanceUrn(String platform, String instance) throws URISyntaxException {
+    return new Urn("urn:li:dataPlatformInstance:(" + new DataPlatformUrn(platform).toString() + "," + instance + ")");
+  }
+
  public static DataFlowUrn flowUrn(String master, String appName) {
    return new DataFlowUrn("spark", appName, master.replaceAll(":", "_").replaceAll("/", "_").replaceAll("[_]+", "_"));


This can be done with a single replaceAll with Regex using a matcher for these three characters

regex way seems more complicated for simple replace. this is more readable.

It's already doing regex though and this does 3 full scans through the String and creates two intermediate Strings that get committed to the String pool as extra memory... I also disagree wholeheartedly that

master.replaceAll("[:\/] | \[\]", "")

is less readable than the current approach, but really don't let this block the PR. It was more of a nitpick in the first place 😄 hopefully the compiler just figures it out anyway.

RyanHolstien · 2022-03-11T19:09:15Z

...ration/java/spark-lineage/src/main/java/datahub/spark/model/dataset/CatalogTableDataset.java

-  public CatalogTableDataset(String dsName) {
-    this.urn = new DatasetUrn(new DataPlatformUrn("hive"), dsName, FabricType.PROD);
+  public CatalogTableDataset(String dsName, String platformInstance, FabricType fabricType) {
+    this.urn = LineageUtils.createDatasetUrn("hive", platformInstance, dsName, fabricType);


This could be pushed up to the interface level with a call to super()

…for spark datasets

shirshanka

LGTM

…datahub-project#4208)

MugdhaHardikar-GSLab added 20 commits January 31, 2022 22:22

fix(spark-lineage): select mock server port dynamically for unit test

09441d3

refactor(spark-lineage): remove dependency of spark from McpEmitter

cac9e4b

fix(spark-lineage): fix checkstyle issues

6f64a20

fix(spark-lineage): restore consumers

8270c24

Merge branch 'linkedin:master' into master

dffa6c4

refactor(spark-lineage): enhance logging and documentation

1675dfe

Update McpEmitter.java

d6583fc

add debugging section

8e9291c

refactor(spark-lineage): enhance logs

7ba4ddf

Merge branch 'spark-logging' of https://github.com/MugdhaHardikar-GSL…

d396664

…ab/datahub into spark-logging

Merge pull request #1 from MugdhaHardikar-GSLab/spark-logging

08b51d7

refactor(spark-lineage): enhance logging and documentation

Add https and auth token support

2c54135

Merge branch 'linkedin:master' into master

7b8ae31

Update README.md

bde1db6

feat(spark-lineage): add support for custom dataset env

bf11982

refactor(spark-lineage): improve jdbc url parsing

d66006e

feat(spark-lineage): add support for pipeline level 'platform_instance'

e8adce2

Update README.md

549f8aa

feat(spark-lineage): support platform instance at dataset level

5169ebf

Update README.md

8044010

fix checkstyle

713e46d

shirshanka requested changes Feb 23, 2022

View reviewed changes

refactor(spark-lineage): change config key name for platform instance

73a695c

shirshanka reviewed Feb 25, 2022

View reviewed changes

RyanHolstien approved these changes Mar 11, 2022

View reviewed changes

refactor(spark-lineage): extract common code at abstract class level …

c82e4fb

…for spark datasets

shirshanka approved these changes Mar 14, 2022

View reviewed changes

shirshanka merged commit a8a4d82 into datahub-project:master Mar 14, 2022

MugdhaHardikar-GSLab deleted the env+platform_instance branch March 30, 2022 05:13

maggiehays pushed a commit to maggiehays/datahub that referenced this pull request Aug 1, 2022

feat(spark-lineage): add support for custom env and platform_instance (…

70d2a52

…datahub-project#4208)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(spark-lineage): add support to custom env and platform_instance #4208

feat(spark-lineage): add support to custom env and platform_instance #4208

MugdhaHardikar-GSLab commented Feb 22, 2022 •

edited

Loading

github-actions bot commented Feb 22, 2022 •

edited

Loading

shirshanka left a comment

shirshanka Feb 23, 2022

MugdhaHardikar-GSLab Feb 23, 2022

shirshanka Feb 23, 2022

MugdhaHardikar-GSLab Feb 23, 2022

shirshanka Feb 23, 2022

MugdhaHardikar-GSLab Feb 23, 2022

shirshanka Feb 23, 2022

MugdhaHardikar-GSLab Feb 23, 2022

shirshanka Feb 23, 2022

MugdhaHardikar-GSLab Feb 23, 2022

RyanHolstien Mar 11, 2022

shirshanka Feb 25, 2022

MugdhaHardikar-GSLab Feb 25, 2022

RyanHolstien left a comment

RyanHolstien Mar 11, 2022

MugdhaHardikar-GSLab Mar 14, 2022

RyanHolstien Mar 14, 2022

RyanHolstien Mar 11, 2022

RyanHolstien Mar 11, 2022

MugdhaHardikar-GSLab Mar 14, 2022

RyanHolstien Mar 11, 2022

MugdhaHardikar-GSLab Mar 14, 2022

RyanHolstien Mar 11, 2022

MugdhaHardikar-GSLab Mar 14, 2022

RyanHolstien Mar 14, 2022

RyanHolstien Mar 11, 2022

MugdhaHardikar-GSLab Mar 14, 2022

shirshanka left a comment

feat(spark-lineage): add support to custom env and platform_instance #4208

feat(spark-lineage): add support to custom env and platform_instance #4208

Conversation

MugdhaHardikar-GSLab commented Feb 22, 2022 • edited Loading

Checklist

github-actions bot commented Feb 22, 2022 • edited Loading

Unit Test Results (build & test)

shirshanka left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

RyanHolstien left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shirshanka left a comment

Choose a reason for hiding this comment

MugdhaHardikar-GSLab commented Feb 22, 2022 •

edited

Loading

github-actions bot commented Feb 22, 2022 •

edited

Loading