You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@kyuubi.apache.org by "wForget (via GitHub)" <gi...@apache.org> on 2023/05/10 12:35:35 UTC
[GitHub] [kyuubi] wForget opened a new pull request, #4815: [KYUUBI #4814] Implements AtlasLineageDispatcher to send lineage to Apache Atlas
wForget opened a new pull request, #4815:
URL: https://github.com/apache/kyuubi/pull/4815
<!--
Thanks for sending a pull request!
Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://kyuubi.readthedocs.io/en/latest/community/CONTRIBUTING.html
2. If the PR is related to an issue in https://github.com/apache/kyuubi/issues, add '[KYUUBI #XXXX]' in your PR title, e.g., '[KYUUBI #XXXX] Your PR title ...'.
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][KYUUBI #XXXX] Your PR title ...'.
-->
### _Why are the changes needed?_
<!--
Please clarify why the changes are needed. For instance,
1. If you add a feature, you can talk about the use case of it.
2. If you fix a bug, you can clarify why it is a bug.
-->
Implements AtlasLineageDispatcher to send lineage to Apache Atlas.
close #4814
### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible
- [ ] Add screenshots for manual tests if appropriate
- [ ] [Run test](https://kyuubi.readthedocs.io/en/master/develop_tools/testing.html#running-tests) locally before make a pull request
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] pan3793 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194772220
##########
extensions/spark/kyuubi-spark-lineage/pom.xml:
##########
@@ -88,7 +87,103 @@
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
- <scope>test</scope>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.jaxrs</groupId>
+ <artifactId>jackson-jaxrs-base</artifactId>
+ <version>${jackson.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.jaxrs</groupId>
+ <artifactId>jackson-jaxrs-json-provider</artifactId>
+ <version>${jackson.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.atlas</groupId>
+ <artifactId>atlas-client-v2</artifactId>
+ <version>${atlas.version}</version>
+ <exclusions>
Review Comment:
what's left after exclusion?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] pan3793 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194769350
##########
extensions/spark/kyuubi-spark-lineage/pom.xml:
##########
@@ -88,7 +87,103 @@
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
- <scope>test</scope>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.jaxrs</groupId>
+ <artifactId>jackson-jaxrs-base</artifactId>
Review Comment:
put compile/provided dependencies before test
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] wForget commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1196258762
##########
extensions/spark/kyuubi-spark-lineage/pom.xml:
##########
@@ -88,7 +87,103 @@
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
- <scope>test</scope>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.jaxrs</groupId>
+ <artifactId>jackson-jaxrs-base</artifactId>
+ <version>${jackson.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.jaxrs</groupId>
+ <artifactId>jackson-jaxrs-json-provider</artifactId>
+ <version>${jackson.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.atlas</groupId>
+ <artifactId>atlas-client-v2</artifactId>
+ <version>${atlas.version}</version>
+ <exclusions>
Review Comment:
full dependencies:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193293040
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -220,11 +220,12 @@ Extra works:
Atlas Client configurations (Configure in `atlas-application.properties` or passed in `spark.atlas.` prefix):
-| Name | Default Value | Description | Since |
-|-----------------------|-----------------|-------------------------------------------------------|-------|
-| atlas.rest.address | localhost:21000 | The rest endpoint url for the Atlas server | 1.8.0 |
-| atlas.client.type | rest | The client type (currently only supports rest) | 1.8.0 |
-| atlas.client.username | none | The client username | 1.8.0 |
-| atlas.client.password | none | The client password | 1.8.0 |
-| atlas.cluster.name | primary | The cluster name to use in qualifiedName of entities. | 1.8.0 |
+| Name | Default Value | Description | Since |
+|------------------------------|-----------------|-------------------------------------------------------|-------|
+| atlas.rest.address | localhost:21000 | The rest endpoint url for the Atlas server | 1.8.0 |
Review Comment:
Whether the protocol should be included as prefix in rest address?
I found all the default values WITH `http` prefix in Atlas's hook is as the following
`atlas.rest.address=http://localhost:31000`
(https://github.com/apache/atlas/blob/master/addons/hive-bridge/src/test/resources/atlas-application.properties#L20)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Implements AtlasLineageDispatcher to send lineage to Apache Atlas
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191828964
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClientConf.scala:
##########
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import org.apache.atlas.ApplicationProperties
+import org.apache.commons.configuration.Configuration
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf.ConfigEntry
+
+class AtlasClientConf(configuration: Configuration) {
+
+ def get(entry: ConfigEntry): String = {
+ configuration.getProperty(entry.key) match {
+ case s: String => s
+ case l: List[_] => l.mkString(",")
+ case o if o != null => o.toString
+ case _ => entry.defaultValue
+ }
+ }
+
+}
+
+object AtlasClientConf {
+
+ private lazy val clientConf: AtlasClientConf = {
+ val conf = ApplicationProperties.get()
+ SparkContextHelper.globalSparkContext.getConf.getAllWithPrefix("spark.atlas.")
+ .foreach {
+ case (k, v) => conf.setProperty(s"atlas.$k", v)
+ }
Review Comment:
```suggestion
.foreach { case (k, v) => conf.setProperty(s"atlas.$k", v) }
```
Minor. In a compact code style.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] codecov-commenter commented on pull request #4815: [KYUUBI #4814] Implements AtlasLineageDispatcher to send lineage to Apache Atlas
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#issuecomment-1542325882
## [Codecov](https://app.codecov.io/gh/apache/kyuubi/pull/4815?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#4815](https://app.codecov.io/gh/apache/kyuubi/pull/4815?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (0aef1be) into [master](https://app.codecov.io/gh/apache/kyuubi/commit/52a11e909beb6d14e098e79323065284c8e34835?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (52a11e9) will **decrease** coverage by `0.09%`.
> The diff coverage is `1.31%`.
```diff
@@ Coverage Diff @@
## master #4815 +/- ##
============================================
- Coverage 58.03% 57.94% -0.09%
Complexity 13 13
============================================
Files 583 587 +4
Lines 32548 32638 +90
Branches 4317 4331 +14
============================================
+ Hits 18889 18912 +23
- Misses 11841 11908 +67
Partials 1818 1818
```
| [Impacted Files](https://app.codecov.io/gh/apache/kyuubi/pull/4815?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...ache/kyuubi/plugin/lineage/LineageDispatcher.scala](https://app.codecov.io/gh/apache/kyuubi/pull/4815?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXh0ZW5zaW9ucy9zcGFyay9reXV1Ymktc3BhcmstbGluZWFnZS9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9wbHVnaW4vbGluZWFnZS9MaW5lYWdlRGlzcGF0Y2hlci5zY2FsYQ==) | `40.00% <0.00%> (-4.45%)` | :arrow_down: |
| [.../plugin/lineage/dispatcher/atlas/AtlasClient.scala](https://app.codecov.io/gh/apache/kyuubi/pull/4815?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXh0ZW5zaW9ucy9zcGFyay9reXV1Ymktc3BhcmstbGluZWFnZS9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9wbHVnaW4vbGluZWFnZS9kaXNwYXRjaGVyL2F0bGFzL0F0bGFzQ2xpZW50LnNjYWxh) | `0.00% <0.00%> (ø)` | |
| [...gin/lineage/dispatcher/atlas/AtlasClientConf.scala](https://app.codecov.io/gh/apache/kyuubi/pull/4815?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXh0ZW5zaW9ucy9zcGFyay9reXV1Ymktc3BhcmstbGluZWFnZS9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9wbHVnaW4vbGluZWFnZS9kaXNwYXRjaGVyL2F0bGFzL0F0bGFzQ2xpZW50Q29uZi5zY2FsYQ==) | `0.00% <0.00%> (ø)` | |
| [...n/lineage/dispatcher/atlas/AtlasEntityHelper.scala](https://app.codecov.io/gh/apache/kyuubi/pull/4815?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXh0ZW5zaW9ucy9zcGFyay9reXV1Ymktc3BhcmstbGluZWFnZS9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9wbHVnaW4vbGluZWFnZS9kaXNwYXRjaGVyL2F0bGFzL0F0bGFzRW50aXR5SGVscGVyLnNjYWxh) | `0.00% <0.00%> (ø)` | |
| [...eage/dispatcher/atlas/AtlasLineageDispatcher.scala](https://app.codecov.io/gh/apache/kyuubi/pull/4815?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXh0ZW5zaW9ucy9zcGFyay9reXV1Ymktc3BhcmstbGluZWFnZS9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9wbHVnaW4vbGluZWFnZS9kaXNwYXRjaGVyL2F0bGFzL0F0bGFzTGluZWFnZURpc3BhdGNoZXIuc2NhbGE=) | `0.00% <0.00%> (ø)` | |
| [.../kyuubi/plugin/lineage/LineageDispatcherType.scala](https://app.codecov.io/gh/apache/kyuubi/pull/4815?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXh0ZW5zaW9ucy9zcGFyay9reXV1Ymktc3BhcmstbGluZWFnZS9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9wbHVnaW4vbGluZWFnZS9MaW5lYWdlRGlzcGF0Y2hlclR5cGUuc2NhbGE=) | `100.00% <100.00%> (ø)` | |
... and [10 files with indirect coverage changes](https://app.codecov.io/gh/apache/kyuubi/pull/4815/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] wForget commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194503468
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -207,3 +208,24 @@ spark.sparkContext.addSparkListener(new SparkListener {
#### Get Lineage Events from Kyuubi EventHandler
When using the `KYUUBI_EVENT` dispatcher, the lineage events will be sent to the Kyuubi `EventBus`. Refer to [Kyuubi Event Handler](../../server/events) to handle kyuubi events.
+
+#### Ingest Lineage Entities to Apache Atlas
+
+The lineage entities can be ingested into [Apache Atlas](https://atlas.apache.org/) using the `ATLAS` dispatcher.
+
+Extra works:
+
++ The least transitive dependencies needed, which are under `./extensions/spark/kyuubi-spark-lineage/target/scala-${scala.binary.version}/jars`
++ Use `spark.files` to specify the `atlas-application.properties` configuration file for Atlas
+
+Atlas Client configurations (Configure in `atlas-application.properties` or passed in `spark.atlas.` prefix):
+
+| Name | Default Value | Description | Since |
+|------------------------------|------------------------|-------------------------------------------------------|-------|
+| atlas.rest.address | http://localhost:21000 | The rest endpoint url for the Atlas server | 1.8.0 |
+| atlas.client.type | rest | The client type (currently only supports rest) | 1.8.0 |
+| atlas.client.username | none | The client username | 1.8.0 |
+| atlas.client.password | none | The client password | 1.8.0 |
+| atlas.cluster.name | primary | The cluster name to use in qualifiedName of entities. | 1.8.0 |
+| atlas.column.lineage.enabled | true | Whether to ingest column lineages to Atlas. | 1.8.0 |
Review Comment:
+1, SGTM
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191924587
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasEntityHelper.scala:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import scala.collection.JavaConverters._
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId, AtlasRelatedObjectId}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper
+
+object AtlasEntityHelper {
+
+ def processEntity(qe: QueryExecution, lineage: Lineage): AtlasEntity = {
+ val entity = new AtlasEntity(PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ val appName = SparkContextHelper.globalSparkContext.appName match {
+ case "Spark shell" => s"Spark Job + $appId"
+ case default => s"$default $appId"
+ }
+
+ entity.setAttribute("qualifiedName", appId)
+ entity.setAttribute("name", appName)
+ entity.setAttribute("currUser", SparkListenerHelper.currentUser)
Review Comment:
LGTM. And maybe you could summarize some key points in the PR description, eg. the entity types, the type model and attributes, the exposed configs, etc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193249797
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -207,3 +208,23 @@ spark.sparkContext.addSparkListener(new SparkListener {
#### Get Lineage Events from Kyuubi EventHandler
When using the `KYUUBI_EVENT` dispatcher, the lineage events will be sent to the Kyuubi `EventBus`. Refer to [Kyuubi Event Handler](../../server/events) to handle kyuubi events.
+
+#### Ingest Lineage Entities to Apache Atlas
+
+The lineage entities can be ingested into [Apache Atlas](https://atlas.apache.org/) using the `ATLAS` dispatcher.
+
+Extra works:
+
++ The least transitive dependencies needed, which are under `./extensions/spark/kyuubi-spark-authz/target/scala-${scala.binary.version}/jars`
Review Comment:
```suggestion
+ The least transitive dependencies needed, which are under `./extensions/spark/kyuubi-spark-lineage/target/scala-${scala.binary.version}/jars`
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] pan3793 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194787377
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClient.scala:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import java.util.Locale
+
+import com.google.common.annotations.VisibleForTesting
+import org.apache.atlas.AtlasClientV2
+import org.apache.atlas.model.instance.AtlasEntity
+import org.apache.atlas.model.instance.AtlasEntity.AtlasEntitiesWithExtInfo
+import org.apache.commons.lang3.StringUtils
+import org.apache.hadoop.util.ShutdownHookManager
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf._
+
+trait AtlasClient {
+ def send(entities: Seq[AtlasEntity]): Unit
+
+ def close(): Unit
+}
+
+class AtlasRestClient(conf: AtlasClientConf) extends AtlasClient {
+
+ private val atlasClient: AtlasClientV2 = {
+ val serverUrl = conf.get(ATLAS_REST_ENDPOINT).split(",")
+ val username = conf.get(CLIENT_USERNAME)
+ val password = conf.get(CLIENT_PASSWORD)
+ if (StringUtils.isNoneBlank(username, password)) {
+ new AtlasClientV2(serverUrl, Array(username, password))
+ } else {
+ new AtlasClientV2(serverUrl: _*)
+ }
+ }
+
+ override def send(entities: Seq[AtlasEntity]): Unit = {
+ val entitiesWithExtInfo = new AtlasEntitiesWithExtInfo()
+ entities.foreach(entitiesWithExtInfo.addEntity)
+ atlasClient.createEntities(entitiesWithExtInfo)
+ }
+
+ override def close(): Unit = {
+ if (atlasClient != null) {
+ atlasClient.close()
+ }
+ }
+}
+
+object AtlasClient {
+
+ @volatile private var client: AtlasClient = _
+
+ def getClient(): AtlasClient = {
+ if (client == null) {
+ AtlasClient.synchronized {
+ if (client == null) {
+ val clientConf = AtlasClientConf.getConf()
+ client = clientConf.get(CLIENT_TYPE).toLowerCase(Locale.ROOT) match {
+ case "rest" => new AtlasRestClient(clientConf)
+ case t => throw new RuntimeException(s"Unsupported client type: $t.")
+ }
+ registerCleanupShutdownHook(client)
+ }
+ }
+ }
+ client
+ }
+
+ private def registerCleanupShutdownHook(client: AtlasClient): Unit = {
+ ShutdownHookManager.get().addShutdownHook(
Review Comment:
eliminate `()` when call a pure function
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] wForget commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194916948
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -207,3 +208,24 @@ spark.sparkContext.addSparkListener(new SparkListener {
#### Get Lineage Events from Kyuubi EventHandler
When using the `KYUUBI_EVENT` dispatcher, the lineage events will be sent to the Kyuubi `EventBus`. Refer to [Kyuubi Event Handler](../../server/events) to handle kyuubi events.
+
+#### Ingest Lineage Entities to Apache Atlas
+
+The lineage entities can be ingested into [Apache Atlas](https://atlas.apache.org/) using the `ATLAS` dispatcher.
+
+Extra works:
+
++ The least transitive dependencies needed, which are under `./extensions/spark/kyuubi-spark-lineage/target/scala-${scala.binary.version}/jars`
++ Use `spark.files` to specify the `atlas-application.properties` configuration file for Atlas
+
+Atlas Client configurations (Configure in `atlas-application.properties` or passed in `spark.atlas.` prefix):
Review Comment:
The `atlas-application.properties` file is required, I just provide the `spark.atlas.` prefix to supplement it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Implements AtlasLineageDispatcher to send lineage to Apache Atlas
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191860968
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasEntityHelper.scala:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import scala.collection.JavaConverters._
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId, AtlasRelatedObjectId}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper
+
+object AtlasEntityHelper {
+
+ def processEntity(qe: QueryExecution, lineage: Lineage): AtlasEntity = {
+ val entity = new AtlasEntity(PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ val appName = SparkContextHelper.globalSparkContext.appName match {
+ case "Spark shell" => s"Spark Job + $appId"
+ case default => s"$default $appId"
+ }
+
+ entity.setAttribute("qualifiedName", appId)
+ entity.setAttribute("name", appName)
+ entity.setAttribute("currUser", SparkListenerHelper.currentUser)
+ entity.setAttribute("executionId", qe.id)
+ entity.setAttribute("details", qe.toString())
+ entity.setAttribute("sparkPlanDescription", qe.sparkPlan.toString())
Review Comment:
We could introduce an entity builder in the followups for entity assembly without repeating the `setAttribute` and `setRelationshipAttribute `. And make the assembling focus in the configs set.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] wForget commented on a diff in pull request #4815: [KYUUBI #4814] Implements AtlasLineageDispatcher to send lineage to Apache Atlas
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191830295
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -207,3 +208,23 @@ spark.sparkContext.addSparkListener(new SparkListener {
#### Get Lineage Events from Kyuubi EventHandler
When using the `KYUUBI_EVENT` dispatcher, the lineage events will be sent to the Kyuubi `EventBus`. Refer to [Kyuubi Event Handler](../../server/events) to handle kyuubi events.
+
+#### Ingest Lineage Entities to Apache Atlas
+
+The lineage entities can be ingested into [Apache Atlas](https://atlas.apache.org/) using the `ATLAS` dispatcher.
+
+Extra works:
+
++ The least transitive dependencies needed, which are under `./extensions/spark/kyuubi-spark-authz/target/scala-${scala.binary.version}/jars`
++ Use `spark.files` to specify the `atlas-application.properties` configuration file for Atlas
+
+Atlas Client configurations (Configure in `atlas-application.properties` or passed in `spark.atlas.` prefix):
+
+| Name | Default Value | Description | Since |
+|-----------------------|-----------------|-------------------------------------------------------|-------|
+| atlas.rest.address | localhost:21000 | The rest endpoint url for the Atlas server | 1.8.0 |
+| atlas.client.type | rest | The client type (currently only supports rest) | 1.8.0 |
+| atlas.client.username | None | The client username | 1.8.0 |
+| atlas.client.password | None | The client password | 1.8.0 |
Review Comment:
> The default value for these two configs should be "" (the same as kyuubi configs), while None is a scala type?
There are also some 'none' in kyuubi docs, like rules.md and settings.md. I prefer to use none, which means no default value instead of an empty string.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] pan3793 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194765943
##########
extensions/spark/kyuubi-spark-lineage/pom.xml:
##########
@@ -54,7 +54,6 @@
<dependency>
<groupId>commons-collections</groupId>
<artifactId>commons-collections</artifactId>
- <scope>test</scope>
Review Comment:
I suppose it should be provided
```
➜ spark-3.3.1-bin-hadoop3 ls jars | grep commons-collections
commons-collections-3.2.2.jar
commons-collections4-4.4.jar
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193293040
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -220,11 +220,12 @@ Extra works:
Atlas Client configurations (Configure in `atlas-application.properties` or passed in `spark.atlas.` prefix):
-| Name | Default Value | Description | Since |
-|-----------------------|-----------------|-------------------------------------------------------|-------|
-| atlas.rest.address | localhost:21000 | The rest endpoint url for the Atlas server | 1.8.0 |
-| atlas.client.type | rest | The client type (currently only supports rest) | 1.8.0 |
-| atlas.client.username | none | The client username | 1.8.0 |
-| atlas.client.password | none | The client password | 1.8.0 |
-| atlas.cluster.name | primary | The cluster name to use in qualifiedName of entities. | 1.8.0 |
+| Name | Default Value | Description | Since |
+|------------------------------|-----------------|-------------------------------------------------------|-------|
+| atlas.rest.address | localhost:21000 | The rest endpoint url for the Atlas server | 1.8.0 |
Review Comment:
Whether the protocol should be included as prefix in value?
I found all the default value in Atlas's hook is as the following
`atlas.rest.address=http://localhost:31000`
(https://github.com/apache/atlas/blob/master/addons/hive-bridge/src/test/resources/atlas-application.properties#L20)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] pan3793 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194782399
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClient.scala:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import java.util.Locale
+
+import com.google.common.annotations.VisibleForTesting
+import org.apache.atlas.AtlasClientV2
+import org.apache.atlas.model.instance.AtlasEntity
+import org.apache.atlas.model.instance.AtlasEntity.AtlasEntitiesWithExtInfo
+import org.apache.commons.lang3.StringUtils
+import org.apache.hadoop.util.ShutdownHookManager
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf._
+
+trait AtlasClient {
+ def send(entities: Seq[AtlasEntity]): Unit
+
+ def close(): Unit
+}
+
+class AtlasRestClient(conf: AtlasClientConf) extends AtlasClient {
+
+ private val atlasClient: AtlasClientV2 = {
+ val serverUrl = conf.get(ATLAS_REST_ENDPOINT).split(",")
+ val username = conf.get(CLIENT_USERNAME)
+ val password = conf.get(CLIENT_PASSWORD)
+ if (StringUtils.isNoneBlank(username, password)) {
+ new AtlasClientV2(serverUrl, Array(username, password))
+ } else {
+ new AtlasClientV2(serverUrl: _*)
+ }
+ }
+
+ override def send(entities: Seq[AtlasEntity]): Unit = {
+ val entitiesWithExtInfo = new AtlasEntitiesWithExtInfo()
+ entities.foreach(entitiesWithExtInfo.addEntity)
+ atlasClient.createEntities(entitiesWithExtInfo)
+ }
+
+ override def close(): Unit = {
+ if (atlasClient != null) {
+ atlasClient.close()
+ }
+ }
+}
+
+object AtlasClient {
+
+ @volatile private var client: AtlasClient = _
Review Comment:
oh, I see the UT overwrite it by a mocked client
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClient.scala:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import java.util.Locale
+
+import com.google.common.annotations.VisibleForTesting
+import org.apache.atlas.AtlasClientV2
+import org.apache.atlas.model.instance.AtlasEntity
+import org.apache.atlas.model.instance.AtlasEntity.AtlasEntitiesWithExtInfo
+import org.apache.commons.lang3.StringUtils
+import org.apache.hadoop.util.ShutdownHookManager
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf._
+
+trait AtlasClient {
+ def send(entities: Seq[AtlasEntity]): Unit
+
+ def close(): Unit
+}
+
+class AtlasRestClient(conf: AtlasClientConf) extends AtlasClient {
+
+ private val atlasClient: AtlasClientV2 = {
+ val serverUrl = conf.get(ATLAS_REST_ENDPOINT).split(",")
+ val username = conf.get(CLIENT_USERNAME)
+ val password = conf.get(CLIENT_PASSWORD)
+ if (StringUtils.isNoneBlank(username, password)) {
+ new AtlasClientV2(serverUrl, Array(username, password))
+ } else {
+ new AtlasClientV2(serverUrl: _*)
+ }
+ }
+
+ override def send(entities: Seq[AtlasEntity]): Unit = {
+ val entitiesWithExtInfo = new AtlasEntitiesWithExtInfo()
+ entities.foreach(entitiesWithExtInfo.addEntity)
+ atlasClient.createEntities(entitiesWithExtInfo)
+ }
+
+ override def close(): Unit = {
+ if (atlasClient != null) {
+ atlasClient.close()
+ }
+ }
+}
+
+object AtlasClient {
+
+ @volatile private var client: AtlasClient = _
Review Comment:
oh, I see the UT overwrites it by a mocked client
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] pan3793 commented on pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#issuecomment-1578312848
Thanks, merged to master for 1.8
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193255960
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasEntityHelper.scala:
##########
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import scala.collection.JavaConverters._
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId, AtlasRelatedObjectId}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper
+
+object AtlasEntityHelper {
+
+ def processEntity(qe: QueryExecution, lineage: Lineage): AtlasEntity = {
+ val entity = new AtlasEntity(PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ val appName = SparkContextHelper.globalSparkContext.appName match {
+ case "Spark shell" => s"Spark Job + $appId"
Review Comment:
```suggestion
case "Spark shell" => s"Spark Job $appId"
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#issuecomment-1547087555
Is the config `atlas.column.lineage.enable` mentioned in docs?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193452388
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -207,3 +208,24 @@ spark.sparkContext.addSparkListener(new SparkListener {
#### Get Lineage Events from Kyuubi EventHandler
When using the `KYUUBI_EVENT` dispatcher, the lineage events will be sent to the Kyuubi `EventBus`. Refer to [Kyuubi Event Handler](../../server/events) to handle kyuubi events.
+
+#### Ingest Lineage Entities to Apache Atlas
+
+The lineage entities can be ingested into [Apache Atlas](https://atlas.apache.org/) using the `ATLAS` dispatcher.
+
+Extra works:
+
++ The least transitive dependencies needed, which are under `./extensions/spark/kyuubi-spark-lineage/target/scala-${scala.binary.version}/jars`
++ Use `spark.files` to specify the `atlas-application.properties` configuration file for Atlas
+
+Atlas Client configurations (Configure in `atlas-application.properties` or passed in `spark.atlas.` prefix):
+
+| Name | Default Value | Description | Since |
+|------------------------------|------------------------|-------------------------------------------------------|-------|
+| atlas.rest.address | http://localhost:21000 | The rest endpoint url for the Atlas server | 1.8.0 |
+| atlas.client.type | rest | The client type (currently only supports rest) | 1.8.0 |
+| atlas.client.username | none | The client username | 1.8.0 |
+| atlas.client.password | none | The client password | 1.8.0 |
+| atlas.cluster.name | primary | The cluster name to use in qualifiedName of entities. | 1.8.0 |
+| atlas.column.lineage.enabled | true | Whether to ingest column lineages to Atlas. | 1.8.0 |
Review Comment:
```suggestion
| atlas.hook.spark.column.lineage.enabled | true | Whether to ingest column lineages to Atlas. | 1.8.0 |
```
This feature uses and serves as the implementation of Atlas's hook. I shall suggest use `atlas.hook.spark.` prefix for the custom configs here. `atlas.hook.` is aligned to the other hooks in Atlas, and the `spark` represents this is an Atlas hook implement for Spark.
WDYT? @wForget @pan3793 @iodone
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] wForget commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191930045
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasEntityHelper.scala:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import scala.collection.JavaConverters._
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId, AtlasRelatedObjectId}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper
+
+object AtlasEntityHelper {
+
+ def processEntity(qe: QueryExecution, lineage: Lineage): AtlasEntity = {
+ val entity = new AtlasEntity(PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ val appName = SparkContextHelper.globalSparkContext.appName match {
+ case "Spark shell" => s"Spark Job + $appId"
+ case default => s"$default $appId"
+ }
+
+ entity.setAttribute("qualifiedName", appId)
+ entity.setAttribute("name", appName)
+ entity.setAttribute("currUser", SparkListenerHelper.currentUser)
+ entity.setAttribute("executionId", qe.id)
+ entity.setAttribute("details", qe.toString())
+ entity.setAttribute("sparkPlanDescription", qe.sparkPlan.toString())
Review Comment:
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Implements AtlasLineageDispatcher to send lineage to Apache Atlas
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191824180
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -207,3 +208,23 @@ spark.sparkContext.addSparkListener(new SparkListener {
#### Get Lineage Events from Kyuubi EventHandler
When using the `KYUUBI_EVENT` dispatcher, the lineage events will be sent to the Kyuubi `EventBus`. Refer to [Kyuubi Event Handler](../../server/events) to handle kyuubi events.
+
+#### Ingest Lineage Entities to Apache Atlas
+
+The lineage entities can be ingested into [Apache Atlas](https://atlas.apache.org/) using the `ATLAS` dispatcher.
+
+Extra works:
+
++ The least transitive dependencies needed, which are under `./extensions/spark/kyuubi-spark-authz/target/scala-${scala.binary.version}/jars`
++ Use `spark.files` to specify the `atlas-application.properties` configuration file for Atlas
+
+Atlas Client configurations (Configure in `atlas-application.properties` or passed in `spark.atlas.` prefix):
+
+| Name | Default Value | Description | Since |
+|-----------------------|-----------------|-------------------------------------------------------|-------|
+| atlas.rest.address | localhost:21000 | The rest endpoint url for the Atlas server | 1.8.0 |
+| atlas.client.type | rest | The client type (currently only supports rest) | 1.8.0 |
+| atlas.client.username | None | The client username | 1.8.0 |
+| atlas.client.password | None | The client password | 1.8.0 |
Review Comment:
The default value for these two configs should be "<undefined>" (the same as kyuubi configs), while None is a scala type?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191871661
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/helper/SparkListenerHelper.scala:
##########
@@ -38,4 +39,6 @@ object SparkListenerHelper {
def isSparkVersionEqualTo(targetVersionString: String): Boolean = {
SemanticVersion(SPARK_VERSION).isVersionEqualTo(targetVersionString)
}
+
+ def currentUser: String = UserGroupInformation.getCurrentUser.getShortUserName
Review Comment:
The current user from UGI may does not reflect the session user. We shall use session property `kyuubi.session.user` as preferred one in the followups.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] wForget commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191921941
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasEntityHelper.scala:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import scala.collection.JavaConverters._
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId, AtlasRelatedObjectId}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper
+
+object AtlasEntityHelper {
+
+ def processEntity(qe: QueryExecution, lineage: Lineage): AtlasEntity = {
+ val entity = new AtlasEntity(PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ val appName = SparkContextHelper.globalSparkContext.appName match {
+ case "Spark shell" => s"Spark Job + $appId"
+ case default => s"$default $appId"
+ }
+
+ entity.setAttribute("qualifiedName", appId)
+ entity.setAttribute("name", appName)
+ entity.setAttribute("currUser", SparkListenerHelper.currentUser)
Review Comment:
> Is the "currUser" attribute a customed one or an existed attribute of any type in Atlas? If not, full name "currentUser" maybe more suitable.
defined in https://github.com/apache/atlas/blob/master/addons/models/1000-Hadoop/1100-spark_model.json
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Implements AtlasLineageDispatcher to send lineage to Apache Atlas
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191809603
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClientConf.scala:
##########
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import org.apache.atlas.ApplicationProperties
+import org.apache.commons.configuration.Configuration
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf.ConfigEntry
+
+class AtlasClientConf(configuration: Configuration) {
+
+ def get(entry: ConfigEntry): String = {
+ configuration.getProperty(entry.key) match {
+ case s: String => s
+ case l: List[_] => l.mkString(",")
+ case o if o != null => o.toString
+ case _ => entry.defaultValue
+ }
+ }
+
+}
+
+object AtlasClientConf {
+
+ private lazy val clientConf: AtlasClientConf = {
+ val conf = ApplicationProperties.get()
+ SparkContextHelper.globalSparkContext.getConf.getAllWithPrefix("spark.atlas.")
+ .foreach {
+ case (k, v) => conf.setProperty(s"atlas.$k", v)
+ }
+ new AtlasClientConf(conf)
+ }
+
+ def getConf(): AtlasClientConf = clientConf
+
+ case class ConfigEntry(key: String, defaultValue: String)
+
+ val ATLAS_REST_ENDPOINT = ConfigEntry("atlas.rest.address", "localhost:21000")
+
+ val CLIENT_TYPE = ConfigEntry("atlas.client.type", "rest")
+ val CLIENT_USERNAME = ConfigEntry("atlas.client.username", null)
+ val CLIENT_PASSWORD = ConfigEntry("atlas.client.password", null)
+
+ val CLUSTER_NAME = ConfigEntry("atlas.cluster.name", "primary")
Review Comment:
SGTM.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193266701
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasEntityHelper.scala:
##########
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import scala.collection.JavaConverters._
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId, AtlasRelatedObjectId}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper
+
+object AtlasEntityHelper {
+
+ def processEntity(qe: QueryExecution, lineage: Lineage): AtlasEntity = {
+ val entity = new AtlasEntity(PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ val appName = SparkContextHelper.globalSparkContext.appName match {
+ case "Spark shell" => s"Spark Job $appId"
+ case default => s"$default $appId"
+ }
+
+ entity.setAttribute("qualifiedName", appId)
+ entity.setAttribute("name", appName)
+ entity.setAttribute("currUser", SparkListenerHelper.currentUser)
+ SparkListenerHelper.sessionUser.foreach(entity.setAttribute("remoteUser", _))
+ entity.setAttribute("executionId", qe.id)
+ entity.setAttribute("details", qe.toString())
+ entity.setAttribute("sparkPlanDescription", qe.sparkPlan.toString())
+
+ // TODO add entity type instead of parsing from string
+ val inputs = lineage.inputTables.flatMap(tableObjectId(_)).map { objId =>
+ relatedObjectId(objId, RELATIONSHIP_DATASET_PROCESS_INPUTS)
+ }
+ val outputs = lineage.outputTables.flatMap(tableObjectId(_)).map { objId =>
+ relatedObjectId(objId, RELATIONSHIP_PROCESS_DATASET_OUTPUTS)
+ }
+
+ entity.setRelationshipAttribute("inputs", inputs.asJava)
+ entity.setRelationshipAttribute("outputs", outputs.asJava)
+
+ entity
+ }
+
+ def columnLineageEntities(processEntity: AtlasEntity, lineage: Lineage): Seq[AtlasEntity] = {
+ lineage.columnLineage.flatMap(columnLineage => {
+ val inputs = columnLineage.originalColumns.flatMap(columnObjectId(_)).map { objId =>
+ relatedObjectId(objId, RELATIONSHIP_DATASET_PROCESS_INPUTS)
+ }
+ val outputs = Option(columnLineage.column).flatMap(columnObjectId(_)).map { objId =>
+ relatedObjectId(objId, RELATIONSHIP_PROCESS_DATASET_OUTPUTS)
+ }.toSeq
+
+ if (!inputs.isEmpty && !outputs.isEmpty) {
Review Comment:
```suggestion
if (inputs.nonEmpty && outputs.nonEmpty) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193259971
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClientConf.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import org.apache.atlas.ApplicationProperties
+import org.apache.commons.configuration.Configuration
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf.ConfigEntry
+
+class AtlasClientConf(configuration: Configuration) {
+
+ def get(entry: ConfigEntry): String = {
+ configuration.getProperty(entry.key) match {
+ case s: String => s
+ case l: List[_] => l.mkString(",")
+ case o if o != null => o.toString
+ case _ => entry.defaultValue
+ }
+ }
+
+}
+
+object AtlasClientConf {
+
+ private lazy val clientConf: AtlasClientConf = {
+ val conf = ApplicationProperties.get()
+ SparkContextHelper.globalSparkContext.getConf.getAllWithPrefix("spark.atlas.")
+ .foreach { case (k, v) => conf.setProperty(s"atlas.$k", v) }
+ new AtlasClientConf(conf)
+ }
+
+ def getConf(): AtlasClientConf = clientConf
+
+ case class ConfigEntry(key: String, defaultValue: String)
+
+ val ATLAS_REST_ENDPOINT = ConfigEntry("atlas.rest.address", "localhost:21000")
+
+ val CLIENT_TYPE = ConfigEntry("atlas.client.type", "rest")
+ val CLIENT_USERNAME = ConfigEntry("atlas.client.username", null)
+ val CLIENT_PASSWORD = ConfigEntry("atlas.client.password", null)
+
+ val CLUSTER_NAME = ConfigEntry("atlas.cluster.name", "primary")
+
+ val COLUMN_LINEAGE_ENABLE = ConfigEntry("atlas.column.lineage.enable", "true")
Review Comment:
```suggestion
val COLUMN_LINEAGE_ENABLED = ConfigEntry("atlas.column.lineage.enabled", true.toString)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193265329
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasEntityHelper.scala:
##########
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import scala.collection.JavaConverters._
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId, AtlasRelatedObjectId}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper
+
+object AtlasEntityHelper {
+
+ def processEntity(qe: QueryExecution, lineage: Lineage): AtlasEntity = {
+ val entity = new AtlasEntity(PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ val appName = SparkContextHelper.globalSparkContext.appName match {
+ case "Spark shell" => s"Spark Job $appId"
+ case default => s"$default $appId"
+ }
+
+ entity.setAttribute("qualifiedName", appId)
+ entity.setAttribute("name", appName)
+ entity.setAttribute("currUser", SparkListenerHelper.currentUser)
+ SparkListenerHelper.sessionUser.foreach(entity.setAttribute("remoteUser", _))
+ entity.setAttribute("executionId", qe.id)
+ entity.setAttribute("details", qe.toString())
+ entity.setAttribute("sparkPlanDescription", qe.sparkPlan.toString())
+
+ // TODO add entity type instead of parsing from string
+ val inputs = lineage.inputTables.flatMap(tableObjectId(_)).map { objId =>
+ relatedObjectId(objId, RELATIONSHIP_DATASET_PROCESS_INPUTS)
+ }
+ val outputs = lineage.outputTables.flatMap(tableObjectId(_)).map { objId =>
Review Comment:
```suggestion
val outputs = lineage.outputTables.flatMap(tableObjectId).map { objId =>
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] pan3793 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194786234
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClient.scala:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import java.util.Locale
+
+import com.google.common.annotations.VisibleForTesting
+import org.apache.atlas.AtlasClientV2
+import org.apache.atlas.model.instance.AtlasEntity
+import org.apache.atlas.model.instance.AtlasEntity.AtlasEntitiesWithExtInfo
+import org.apache.commons.lang3.StringUtils
+import org.apache.hadoop.util.ShutdownHookManager
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf._
+
+trait AtlasClient {
+ def send(entities: Seq[AtlasEntity]): Unit
+
+ def close(): Unit
+}
+
+class AtlasRestClient(conf: AtlasClientConf) extends AtlasClient {
+
+ private val atlasClient: AtlasClientV2 = {
+ val serverUrl = conf.get(ATLAS_REST_ENDPOINT).split(",")
+ val username = conf.get(CLIENT_USERNAME)
+ val password = conf.get(CLIENT_PASSWORD)
+ if (StringUtils.isNoneBlank(username, password)) {
+ new AtlasClientV2(serverUrl, Array(username, password))
+ } else {
+ new AtlasClientV2(serverUrl: _*)
+ }
+ }
+
+ override def send(entities: Seq[AtlasEntity]): Unit = {
+ val entitiesWithExtInfo = new AtlasEntitiesWithExtInfo()
+ entities.foreach(entitiesWithExtInfo.addEntity)
+ atlasClient.createEntities(entitiesWithExtInfo)
+ }
+
+ override def close(): Unit = {
+ if (atlasClient != null) {
+ atlasClient.close()
+ }
+ }
+}
+
+object AtlasClient {
+
+ @volatile private var client: AtlasClient = _
+
+ def getClient(): AtlasClient = {
+ if (client == null) {
+ AtlasClient.synchronized {
+ if (client == null) {
+ val clientConf = AtlasClientConf.getConf()
+ client = clientConf.get(CLIENT_TYPE).toLowerCase(Locale.ROOT) match {
+ case "rest" => new AtlasRestClient(clientConf)
+ case t => throw new RuntimeException(s"Unsupported client type: $t.")
Review Comment:
```suggestion
case unknown => throw new RuntimeException(s"Unsupported client type: $unknown.")
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] pan3793 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194791235
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClientConf.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import org.apache.atlas.ApplicationProperties
+import org.apache.commons.configuration.Configuration
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf.ConfigEntry
+
+class AtlasClientConf(configuration: Configuration) {
+
+ def get(entry: ConfigEntry): String = {
+ configuration.getProperty(entry.key) match {
+ case s: String => s
+ case l: List[_] => l.mkString(",")
+ case o if o != null => o.toString
+ case _ => entry.defaultValue
+ }
+ }
+
+}
+
+object AtlasClientConf {
+
+ private lazy val clientConf: AtlasClientConf = {
+ val conf = ApplicationProperties.get()
+ SparkContextHelper.globalSparkContext.getConf.getAllWithPrefix("spark.atlas.")
+ .foreach { case (k, v) => conf.setProperty(s"atlas.$k", v) }
+ new AtlasClientConf(conf)
+ }
+
+ def getConf(): AtlasClientConf = clientConf
+
+ case class ConfigEntry(key: String, defaultValue: String)
Review Comment:
lift to the package, it's weird defining a class inside a companion object which would be used outside
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] wForget commented on a diff in pull request #4815: [KYUUBI #4814] Implements AtlasLineageDispatcher to send lineage to Apache Atlas
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191822410
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClient.scala:
##########
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import java.util.Locale
+
+import com.google.common.annotations.VisibleForTesting
+import org.apache.atlas.AtlasClientV2
+import org.apache.atlas.model.instance.AtlasEntity
+import org.apache.atlas.model.instance.AtlasEntity.AtlasEntitiesWithExtInfo
+import org.apache.commons.lang3.StringUtils
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf._
+
+trait AtlasClient {
+ def send(entities: Seq[AtlasEntity]): Unit
+}
+
+class AtlasRestClient(conf: AtlasClientConf) extends AtlasClient {
+
+ private val atlasClient: AtlasClientV2 = {
+ val serverUrl = conf.get(ATLAS_REST_ENDPOINT).split(",")
+ val username = conf.get(CLIENT_USERNAME)
+ val password = conf.get(CLIENT_PASSWORD)
+ if (StringUtils.isNotBlank(username) && StringUtils.isNotBlank(password)) {
Review Comment:
thanks, fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] wForget commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191928129
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/helper/SparkListenerHelper.scala:
##########
@@ -38,4 +39,6 @@ object SparkListenerHelper {
def isSparkVersionEqualTo(targetVersionString: String): Boolean = {
SemanticVersion(SPARK_VERSION).isVersionEqualTo(targetVersionString)
}
+
+ def currentUser: String = UserGroupInformation.getCurrentUser.getShortUserName
Review Comment:
> The current user from UGI may does not reflect the session user. We shall use session property ""kyuubi.session.user" as preferred one in the followups.
Thanks, I didn't find a suitable way to get the session user before. But, I think it is more appropriate to be set to `remoteUser`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on pull request #4815: [KYUUBI #4814] Implements AtlasLineageDispatcher to send lineage to Apache Atlas
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#issuecomment-1543572469
Nice work. Good to have Atalas hook support in the lineage plugin!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Implements AtlasLineageDispatcher to send lineage to Apache Atlas
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1190859851
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClientConf.scala:
##########
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import org.apache.atlas.ApplicationProperties
+import org.apache.commons.configuration.Configuration
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf.ConfigEntry
+
+class AtlasClientConf(configuration: Configuration) {
+
+ def get(entry: ConfigEntry): String = {
+ configuration.getProperty(entry.key) match {
+ case s: String => s
+ case l: List[_] => l.mkString(",")
+ case o if o != null => o.toString
+ case _ => entry.defaultValue
+ }
+ }
+
+}
+
+object AtlasClientConf {
+
+ private lazy val clientConf: AtlasClientConf = {
+ val conf = ApplicationProperties.get()
+ SparkContextHelper.globalSparkContext.getConf.getAllWithPrefix("spark.atlas.")
+ .foreach {
+ case (k, v) => conf.setProperty(s"atlas.$k", v)
+ }
+ new AtlasClientConf(conf)
+ }
+
+ def getConf(): AtlasClientConf = clientConf
+
+ case class ConfigEntry(key: String, defaultValue: String)
+
+ val ATLAS_REST_ENDPOINT = ConfigEntry("atlas.rest.address", "localhost:21000")
+
+ val CLIENT_TYPE = ConfigEntry("atlas.client.type", "rest")
+ val CLIENT_USERNAME = ConfigEntry("atlas.client.username", null)
+ val CLIENT_PASSWORD = ConfigEntry("atlas.client.password", null)
+
+ val CLUSTER_NAME = ConfigEntry("atlas.cluster.name", "primary")
Review Comment:
I'm not sure any default value should be provided to this config. The cluster name would be leave undefined by default but to check it is required.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] pan3793 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194763318
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -207,3 +208,24 @@ spark.sparkContext.addSparkListener(new SparkListener {
#### Get Lineage Events from Kyuubi EventHandler
When using the `KYUUBI_EVENT` dispatcher, the lineage events will be sent to the Kyuubi `EventBus`. Refer to [Kyuubi Event Handler](../../server/events) to handle kyuubi events.
+
+#### Ingest Lineage Entities to Apache Atlas
+
+The lineage entities can be ingested into [Apache Atlas](https://atlas.apache.org/) using the `ATLAS` dispatcher.
+
+Extra works:
+
++ The least transitive dependencies needed, which are under `./extensions/spark/kyuubi-spark-lineage/target/scala-${scala.binary.version}/jars`
++ Use `spark.files` to specify the `atlas-application.properties` configuration file for Atlas
+
+Atlas Client configurations (Configure in `atlas-application.properties` or passed in `spark.atlas.` prefix):
Review Comment:
these two sentences are not consistent, BTW, does it work to put `atlas-application.properties` under `SPARK_CONF_DIR`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193260900
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasLineageDispatcher.scala:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.{Lineage, LineageDispatcher}
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasEntityHelper.columnLineageEnable
+
+class AtlasLineageDispatcher extends LineageDispatcher with Logging {
+
+ override def send(qe: QueryExecution, lineageOpt: Option[Lineage]): Unit = {
+ try {
+ lineageOpt.filter(l => l.inputTables.nonEmpty || l.outputTables.nonEmpty).foreach(lineage => {
+ val processEntity = AtlasEntityHelper.processEntity(qe, lineage)
+ val columnLineageEntities = if (lineage.columnLineage.nonEmpty && columnLineageEnable) {
Review Comment:
```suggestion
val columnLineageEntities = if (lineage.columnLineage.nonEmpty && columnLineageEnabled) {
```
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClientConf.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import org.apache.atlas.ApplicationProperties
+import org.apache.commons.configuration.Configuration
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf.ConfigEntry
+
+class AtlasClientConf(configuration: Configuration) {
+
+ def get(entry: ConfigEntry): String = {
+ configuration.getProperty(entry.key) match {
+ case s: String => s
+ case l: List[_] => l.mkString(",")
+ case o if o != null => o.toString
+ case _ => entry.defaultValue
+ }
+ }
+
+}
+
+object AtlasClientConf {
+
+ private lazy val clientConf: AtlasClientConf = {
+ val conf = ApplicationProperties.get()
+ SparkContextHelper.globalSparkContext.getConf.getAllWithPrefix("spark.atlas.")
+ .foreach { case (k, v) => conf.setProperty(s"atlas.$k", v) }
+ new AtlasClientConf(conf)
+ }
+
+ def getConf(): AtlasClientConf = clientConf
+
+ case class ConfigEntry(key: String, defaultValue: String)
+
+ val ATLAS_REST_ENDPOINT = ConfigEntry("atlas.rest.address", "localhost:21000")
+
+ val CLIENT_TYPE = ConfigEntry("atlas.client.type", "rest")
+ val CLIENT_USERNAME = ConfigEntry("atlas.client.username", null)
+ val CLIENT_PASSWORD = ConfigEntry("atlas.client.password", null)
+
+ val CLUSTER_NAME = ConfigEntry("atlas.cluster.name", "primary")
+
+ val COLUMN_LINEAGE_ENABLE = ConfigEntry("atlas.column.lineage.enable", "true")
Review Comment:
```suggestion
val COLUMN_LINEAGE_ENABLED = ConfigEntry("atlas.column.lineage.enabled", "true")
```
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasEntityHelper.scala:
##########
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import scala.collection.JavaConverters._
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId, AtlasRelatedObjectId}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper
+
+object AtlasEntityHelper {
+
+ def processEntity(qe: QueryExecution, lineage: Lineage): AtlasEntity = {
+ val entity = new AtlasEntity(PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ val appName = SparkContextHelper.globalSparkContext.appName match {
+ case "Spark shell" => s"Spark Job $appId"
+ case default => s"$default $appId"
+ }
+
+ entity.setAttribute("qualifiedName", appId)
+ entity.setAttribute("name", appName)
+ entity.setAttribute("currUser", SparkListenerHelper.currentUser)
+ SparkListenerHelper.sessionUser.foreach(entity.setAttribute("remoteUser", _))
+ entity.setAttribute("executionId", qe.id)
+ entity.setAttribute("details", qe.toString())
+ entity.setAttribute("sparkPlanDescription", qe.sparkPlan.toString())
+
+ // TODO add entity type instead of parsing from string
+ val inputs = lineage.inputTables.flatMap(tableObjectId(_)).map { objId =>
+ relatedObjectId(objId, RELATIONSHIP_DATASET_PROCESS_INPUTS)
+ }
+ val outputs = lineage.outputTables.flatMap(tableObjectId(_)).map { objId =>
+ relatedObjectId(objId, RELATIONSHIP_PROCESS_DATASET_OUTPUTS)
+ }
+
+ entity.setRelationshipAttribute("inputs", inputs.asJava)
+ entity.setRelationshipAttribute("outputs", outputs.asJava)
+
+ entity
+ }
+
+ def columnLineageEntities(processEntity: AtlasEntity, lineage: Lineage): Seq[AtlasEntity] = {
+ lineage.columnLineage.flatMap(columnLineage => {
+ val inputs = columnLineage.originalColumns.flatMap(columnObjectId(_)).map { objId =>
+ relatedObjectId(objId, RELATIONSHIP_DATASET_PROCESS_INPUTS)
+ }
+ val outputs = Option(columnLineage.column).flatMap(columnObjectId(_)).map { objId =>
+ relatedObjectId(objId, RELATIONSHIP_PROCESS_DATASET_OUTPUTS)
+ }.toSeq
+
+ if (!inputs.isEmpty && !outputs.isEmpty) {
+ val entity = new AtlasEntity(COLUMN_LINEAGE_TYPE)
+ val qualifiedName =
+ s"${processEntity.getAttribute("qualifiedName")}:${columnLineage.column}"
+ entity.setAttribute("qualifiedName", qualifiedName)
+ entity.setAttribute("name", qualifiedName)
+ entity.setRelationshipAttribute("inputs", inputs.asJava)
+ entity.setRelationshipAttribute("outputs", outputs.asJava)
+ entity.setRelationshipAttribute(
+ "process",
+ relatedObjectId(objectId(processEntity), RELATIONSHIP_SPARK_PROCESS_COLUMN_LINEAGE))
+ Some(entity)
+ } else {
+ None
+ }
+ })
+ }
+
+ def tableObjectId(tableName: String): Option[AtlasObjectId] = {
+ val dbTb = tableName.split('.')
+ if (dbTb.length == 2) {
+ val qualifiedName = tableQualifiedName(cluster, dbTb(0), dbTb(1))
+ // TODO parse datasource type
+ Some(new AtlasObjectId(HIVE_TABLE_TYPE, "qualifiedName", qualifiedName))
+ } else {
+ None
+ }
+ }
+
+ def tableQualifiedName(cluster: String, db: String, table: String): String = {
+ s"${db.toLowerCase}.${table.toLowerCase}@$cluster"
+ }
+
+ def columnObjectId(columnName: String): Option[AtlasObjectId] = {
+ val dbTbCol = columnName.split('.')
+ if (dbTbCol.length == 3) {
+ val qualifiedName = columnQualifiedName(cluster, dbTbCol(0), dbTbCol(1), dbTbCol(2))
+ // TODO parse datasource type
+ Some(new AtlasObjectId(HIVE_COLUMN_TYPE, "qualifiedName", qualifiedName))
+ } else {
+ None
+ }
+ }
+
+ def columnQualifiedName(cluster: String, db: String, table: String, column: String): String = {
+ s"${db.toLowerCase}.${table.toLowerCase}.${column.toLowerCase}@$cluster"
+ }
+
+ def objectId(entity: AtlasEntity): AtlasObjectId = {
+ val objId = new AtlasObjectId(entity.getGuid, entity.getTypeName)
+ objId.setUniqueAttributes(Map("qualifiedName" -> entity.getAttribute("qualifiedName")).asJava)
+ objId
+ }
+
+ def relatedObjectId(objectId: AtlasObjectId, relationshipType: String): AtlasRelatedObjectId = {
+ new AtlasRelatedObjectId(objectId, relationshipType)
+ }
+
+ lazy val cluster = AtlasClientConf.getConf().get(AtlasClientConf.CLUSTER_NAME)
+ lazy val columnLineageEnable =
+ AtlasClientConf.getConf().get(AtlasClientConf.COLUMN_LINEAGE_ENABLE).toBoolean
Review Comment:
```suggestion
lazy val columnLineageEnabled =
AtlasClientConf.getConf().get(AtlasClientConf.COLUMN_LINEAGE_ENABLED).toBoolean
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193293817
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -207,3 +208,24 @@ spark.sparkContext.addSparkListener(new SparkListener {
#### Get Lineage Events from Kyuubi EventHandler
When using the `KYUUBI_EVENT` dispatcher, the lineage events will be sent to the Kyuubi `EventBus`. Refer to [Kyuubi Event Handler](../../server/events) to handle kyuubi events.
+
+#### Ingest Lineage Entities to Apache Atlas
+
+The lineage entities can be ingested into [Apache Atlas](https://atlas.apache.org/) using the `ATLAS` dispatcher.
+
+Extra works:
+
++ The least transitive dependencies needed, which are under `./extensions/spark/kyuubi-spark-lineage/target/scala-${scala.binary.version}/jars`
++ Use `spark.files` to specify the `atlas-application.properties` configuration file for Atlas
+
+Atlas Client configurations (Configure in `atlas-application.properties` or passed in `spark.atlas.` prefix):
+
+| Name | Default Value | Description | Since |
+|------------------------------|-----------------|-------------------------------------------------------|-------|
+| atlas.rest.address | localhost:21000 | The rest endpoint url for the Atlas server | 1.8.0 |
Review Comment:
Whether the protocol should be included as prefix in rest address?
I found all the default values WITH `http` prefix in Atlas's hook is as the following
`atlas.rest.address=http://localhost:31000`
(https://github.com/apache/atlas/blob/master/addons/hive-bridge/src/test/resources/atlas-application.properties#L20)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] pan3793 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194776006
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClient.scala:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import java.util.Locale
+
+import com.google.common.annotations.VisibleForTesting
+import org.apache.atlas.AtlasClientV2
+import org.apache.atlas.model.instance.AtlasEntity
+import org.apache.atlas.model.instance.AtlasEntity.AtlasEntitiesWithExtInfo
+import org.apache.commons.lang3.StringUtils
+import org.apache.hadoop.util.ShutdownHookManager
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf._
+
+trait AtlasClient {
+ def send(entities: Seq[AtlasEntity]): Unit
+
+ def close(): Unit
Review Comment:
how about `extends AutoClosable`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193273130
##########
extensions/spark/kyuubi-spark-lineage/src/test/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasLineageDispatcherSuite.scala:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import java.util
+
+import scala.collection.JavaConverters._
+import scala.collection.immutable.List
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId}
+import org.apache.commons.lang3.StringUtils
+import org.apache.spark.SparkConf
+import org.apache.spark.kyuubi.lineage.LineageConf.{DISPATCHERS, SKIP_PARSING_PERMANENT_VIEW_ENABLED}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.SparkListenerExtensionTest
+import org.scalatest.concurrent.PatienceConfiguration.Timeout
+import org.scalatest.time.SpanSugar._
+
+import org.apache.kyuubi.KyuubiFunSuite
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasEntityHelper.{COLUMN_LINEAGE_TYPE, PROCESS_TYPE}
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper.isSparkVersionAtMost
+
+class AtlasLineageDispatcherSuite extends KyuubiFunSuite with SparkListenerExtensionTest {
+ val catalogName =
+ if (isSparkVersionAtMost("3.1")) "org.apache.spark.sql.connector.InMemoryTableCatalog"
+ else "org.apache.spark.sql.connector.catalog.InMemoryTableCatalog"
+
+ override protected val catalogImpl: String = "hive"
+
+ override def sparkConf(): SparkConf = {
+ super.sparkConf()
+ .set("spark.sql.catalog.v2_catalog", catalogName)
+ .set(
+ "spark.sql.queryExecutionListeners",
+ "org.apache.kyuubi.plugin.lineage.SparkOperationLineageQueryExecutionListener")
+ .set(DISPATCHERS.key, "ATLAS")
+ .set(SKIP_PARSING_PERMANENT_VIEW_ENABLED.key, "true")
+ }
+
+ override def afterAll(): Unit = {
+ spark.stop()
+ super.afterAll()
+ }
+
+ test("altas lineage capture: insert into select sql") {
+ val mockAtlasClient = new MockAtlasClient()
+ AtlasClient.setClient(mockAtlasClient)
+
+ withTable("test_table0") { _ =>
+ spark.sql("create table test_table0(a string, b int, c int)")
+ spark.sql("create table test_table1(a string, d int)")
+ spark.sql("insert into test_table1 select a, b + c as d from test_table0").collect()
+ val expected = Lineage(
+ List("default.test_table0"),
+ List("default.test_table1"),
+ List(
+ ("default.test_table1.a", Set("default.test_table0.a")),
+ ("default.test_table1.d", Set("default.test_table0.b", "default.test_table0.c"))))
+ eventually(Timeout(5.seconds)) {
+ assert(mockAtlasClient.getEntities != null && mockAtlasClient.getEntities.nonEmpty)
+ }
+ checkAtlasProcessEntity(mockAtlasClient.getEntities.head, expected)
+ checkAtlasColumnLineageEntities(
+ mockAtlasClient.getEntities.head,
+ mockAtlasClient.getEntities.tail,
+ expected)
+ }
+
+ }
+
+ def checkAtlasProcessEntity(entity: AtlasEntity, expected: Lineage): Unit = {
+ assert(entity.getTypeName == PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ assert(entity.getAttribute("qualifiedName") == appId)
+ assert(entity.getAttribute("name")
+ == s"${SparkContextHelper.globalSparkContext.appName} $appId")
+ assert(StringUtils.isNotBlank(entity.getAttribute("currUser").asInstanceOf[String]))
+ assert(entity.getAttribute("executionId") != null)
+ assert(StringUtils.isNotBlank(entity.getAttribute("details").asInstanceOf[String]))
+ assert(StringUtils.isNotBlank(entity.getAttribute("sparkPlanDescription").asInstanceOf[String]))
+
+ val inputs = entity.getRelationshipAttribute("inputs")
+ .asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName)
+ val outputs = entity.getRelationshipAttribute("outputs")
+ .asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName)
+ assertResult(expected.inputTables.map(s => s"$s@$cluster"))(inputs)
+ assertResult(expected.outputTables.map(s => s"$s@$cluster"))(outputs)
+ }
+
+ def checkAtlasColumnLineageEntities(
+ processEntity: AtlasEntity,
+ entities: Seq[AtlasEntity],
+ expected: Lineage): Unit = {
+ assert(entities.size == expected.columnLineage.size)
+
+ entities.zip(expected.columnLineage).foreach {
+ case (entity, expectedLineage) =>
+ assert(entity.getTypeName == COLUMN_LINEAGE_TYPE)
+ val expectedQualifiedName =
+ s"${processEntity.getAttribute("qualifiedName")}:${expectedLineage.column}"
+ assert(entity.getAttribute("qualifiedName") == expectedQualifiedName)
+ assert(entity.getAttribute("name") == expectedQualifiedName)
+
+ val inputs = entity.getRelationshipAttribute("inputs")
+ .asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName)
+ assertResult(expectedLineage.originalColumns.map(s => s"$s@$cluster"))(inputs.toSet)
+
+ val outputs = entity.getRelationshipAttribute("outputs")
+ .asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName)
+ assert(outputs.size == 1)
+ assert(s"${expectedLineage.column}@$cluster" == outputs.head)
+
+ assert(getQualifiedName(entity.getRelationshipAttribute("process").asInstanceOf[
+ AtlasObjectId]) == processEntity.getAttribute("qualifiedName"))
+ }
+ }
+
+ // Configured in atlas-application.properties
+ val cluster = "test"
Review Comment:
```suggestion
// Pre-set cluster name for testing in `test/resources/atlas-application.properties`
private val cluster = "test"
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] wForget commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193251747
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -207,3 +208,23 @@ spark.sparkContext.addSparkListener(new SparkListener {
#### Get Lineage Events from Kyuubi EventHandler
When using the `KYUUBI_EVENT` dispatcher, the lineage events will be sent to the Kyuubi `EventBus`. Refer to [Kyuubi Event Handler](../../server/events) to handle kyuubi events.
+
+#### Ingest Lineage Entities to Apache Atlas
+
+The lineage entities can be ingested into [Apache Atlas](https://atlas.apache.org/) using the `ATLAS` dispatcher.
+
+Extra works:
+
++ The least transitive dependencies needed, which are under `./extensions/spark/kyuubi-spark-authz/target/scala-${scala.binary.version}/jars`
Review Comment:
thanks, fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194934892
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClient.scala:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import java.util.Locale
+
+import com.google.common.annotations.VisibleForTesting
+import org.apache.atlas.AtlasClientV2
+import org.apache.atlas.model.instance.AtlasEntity
+import org.apache.atlas.model.instance.AtlasEntity.AtlasEntitiesWithExtInfo
+import org.apache.commons.lang3.StringUtils
+import org.apache.hadoop.util.ShutdownHookManager
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf._
+
+trait AtlasClient {
+ def send(entities: Seq[AtlasEntity]): Unit
+
+ def close(): Unit
+}
+
+class AtlasRestClient(conf: AtlasClientConf) extends AtlasClient {
+
+ private val atlasClient: AtlasClientV2 = {
+ val serverUrl = conf.get(ATLAS_REST_ENDPOINT).split(",")
+ val username = conf.get(CLIENT_USERNAME)
+ val password = conf.get(CLIENT_PASSWORD)
+ if (StringUtils.isNoneBlank(username, password)) {
+ new AtlasClientV2(serverUrl, Array(username, password))
+ } else {
+ new AtlasClientV2(serverUrl: _*)
+ }
+ }
+
+ override def send(entities: Seq[AtlasEntity]): Unit = {
+ val entitiesWithExtInfo = new AtlasEntitiesWithExtInfo()
+ entities.foreach(entitiesWithExtInfo.addEntity)
+ atlasClient.createEntities(entitiesWithExtInfo)
+ }
+
+ override def close(): Unit = {
+ if (atlasClient != null) {
+ atlasClient.close()
+ }
+ }
+}
+
+object AtlasClient {
+
+ @volatile private var client: AtlasClient = _
+
+ def getClient(): AtlasClient = {
+ if (client == null) {
+ AtlasClient.synchronized {
+ if (client == null) {
+ val clientConf = AtlasClientConf.getConf()
+ client = clientConf.get(CLIENT_TYPE).toLowerCase(Locale.ROOT) match {
+ case "rest" => new AtlasRestClient(clientConf)
+ case t => throw new RuntimeException(s"Unsupported client type: $t.")
+ }
+ registerCleanupShutdownHook(client)
+ }
+ }
+ }
+ client
+ }
+
+ private def registerCleanupShutdownHook(client: AtlasClient): Unit = {
+ ShutdownHookManager.get().addShutdownHook(
Review Comment:
```
ShutdownHookManager.get.addShutdownHook(
() => Option(client).foreach(_.close()),
Integer.MAX_VALUE)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194934892
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClient.scala:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import java.util.Locale
+
+import com.google.common.annotations.VisibleForTesting
+import org.apache.atlas.AtlasClientV2
+import org.apache.atlas.model.instance.AtlasEntity
+import org.apache.atlas.model.instance.AtlasEntity.AtlasEntitiesWithExtInfo
+import org.apache.commons.lang3.StringUtils
+import org.apache.hadoop.util.ShutdownHookManager
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf._
+
+trait AtlasClient {
+ def send(entities: Seq[AtlasEntity]): Unit
+
+ def close(): Unit
+}
+
+class AtlasRestClient(conf: AtlasClientConf) extends AtlasClient {
+
+ private val atlasClient: AtlasClientV2 = {
+ val serverUrl = conf.get(ATLAS_REST_ENDPOINT).split(",")
+ val username = conf.get(CLIENT_USERNAME)
+ val password = conf.get(CLIENT_PASSWORD)
+ if (StringUtils.isNoneBlank(username, password)) {
+ new AtlasClientV2(serverUrl, Array(username, password))
+ } else {
+ new AtlasClientV2(serverUrl: _*)
+ }
+ }
+
+ override def send(entities: Seq[AtlasEntity]): Unit = {
+ val entitiesWithExtInfo = new AtlasEntitiesWithExtInfo()
+ entities.foreach(entitiesWithExtInfo.addEntity)
+ atlasClient.createEntities(entitiesWithExtInfo)
+ }
+
+ override def close(): Unit = {
+ if (atlasClient != null) {
+ atlasClient.close()
+ }
+ }
+}
+
+object AtlasClient {
+
+ @volatile private var client: AtlasClient = _
+
+ def getClient(): AtlasClient = {
+ if (client == null) {
+ AtlasClient.synchronized {
+ if (client == null) {
+ val clientConf = AtlasClientConf.getConf()
+ client = clientConf.get(CLIENT_TYPE).toLowerCase(Locale.ROOT) match {
+ case "rest" => new AtlasRestClient(clientConf)
+ case t => throw new RuntimeException(s"Unsupported client type: $t.")
+ }
+ registerCleanupShutdownHook(client)
+ }
+ }
+ }
+ client
+ }
+
+ private def registerCleanupShutdownHook(client: AtlasClient): Unit = {
+ ShutdownHookManager.get().addShutdownHook(
Review Comment:
```
ShutdownHookManager.get.addShutdownHook(
() => Option(client).foreach(_.close()),
Integer.MAX_VALUE)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] pan3793 closed pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 closed pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
URL: https://github.com/apache/kyuubi/pull/4815
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] wForget commented on a diff in pull request #4815: [KYUUBI #4814] Implements AtlasLineageDispatcher to send lineage to Apache Atlas
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1190882487
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClientConf.scala:
##########
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import org.apache.atlas.ApplicationProperties
+import org.apache.commons.configuration.Configuration
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf.ConfigEntry
+
+class AtlasClientConf(configuration: Configuration) {
+
+ def get(entry: ConfigEntry): String = {
+ configuration.getProperty(entry.key) match {
+ case s: String => s
+ case l: List[_] => l.mkString(",")
+ case o if o != null => o.toString
+ case _ => entry.defaultValue
+ }
+ }
+
+}
+
+object AtlasClientConf {
+
+ private lazy val clientConf: AtlasClientConf = {
+ val conf = ApplicationProperties.get()
+ SparkContextHelper.globalSparkContext.getConf.getAllWithPrefix("spark.atlas.")
+ .foreach {
+ case (k, v) => conf.setProperty(s"atlas.$k", v)
+ }
+ new AtlasClientConf(conf)
+ }
+
+ def getConf(): AtlasClientConf = clientConf
+
+ case class ConfigEntry(key: String, defaultValue: String)
+
+ val ATLAS_REST_ENDPOINT = ConfigEntry("atlas.rest.address", "localhost:21000")
+
+ val CLIENT_TYPE = ConfigEntry("atlas.client.type", "rest")
+ val CLIENT_USERNAME = ConfigEntry("atlas.client.username", null)
+ val CLIENT_PASSWORD = ConfigEntry("atlas.client.password", null)
+
+ val CLUSTER_NAME = ConfigEntry("atlas.cluster.name", "primary")
Review Comment:
> I'm not sure any default value should be provided to this config. The cluster name would be left null by default but to check it is required.
The default value of atlas.cluster.name in hive hook is primary, we need to be consistent with it.
FYI: https://atlas.apache.org/#/HookHive
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191870772
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasEntityHelper.scala:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import scala.collection.JavaConverters._
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId, AtlasRelatedObjectId}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper
+
+object AtlasEntityHelper {
+
+ def processEntity(qe: QueryExecution, lineage: Lineage): AtlasEntity = {
+ val entity = new AtlasEntity(PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ val appName = SparkContextHelper.globalSparkContext.appName match {
+ case "Spark shell" => s"Spark Job + $appId"
+ case default => s"$default $appId"
+ }
+
+ entity.setAttribute("qualifiedName", appId)
+ entity.setAttribute("name", appName)
+ entity.setAttribute("currUser", SparkListenerHelper.currentUser)
Review Comment:
Is the "currUser" attribute a customed one or an existed attribute of any type in Atlas? If not, full name "currentUser" maybe more suitable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191871661
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/helper/SparkListenerHelper.scala:
##########
@@ -38,4 +39,6 @@ object SparkListenerHelper {
def isSparkVersionEqualTo(targetVersionString: String): Boolean = {
SemanticVersion(SPARK_VERSION).isVersionEqualTo(targetVersionString)
}
+
+ def currentUser: String = UserGroupInformation.getCurrentUser.getShortUserName
Review Comment:
The current user from UGI may does not reflect the session user. We shall use session property ""kyuubi.session.user" as preferred one in the followups.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] wForget commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191924084
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasEntityHelper.scala:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import scala.collection.JavaConverters._
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId, AtlasRelatedObjectId}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper
+
+object AtlasEntityHelper {
+
+ def processEntity(qe: QueryExecution, lineage: Lineage): AtlasEntity = {
+ val entity = new AtlasEntity(PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ val appName = SparkContextHelper.globalSparkContext.appName match {
+ case "Spark shell" => s"Spark Job + $appId"
+ case default => s"$default $appId"
+ }
+
+ entity.setAttribute("qualifiedName", appId)
+ entity.setAttribute("name", appName)
Review Comment:
> We could use session configs for setting appId and appName in the followups, allowing user to custom the key details of the process entities.
SGTM, we can implement it in follow-up pr.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Implements AtlasLineageDispatcher to send lineage to Apache Atlas
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191828964
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClientConf.scala:
##########
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import org.apache.atlas.ApplicationProperties
+import org.apache.commons.configuration.Configuration
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf.ConfigEntry
+
+class AtlasClientConf(configuration: Configuration) {
+
+ def get(entry: ConfigEntry): String = {
+ configuration.getProperty(entry.key) match {
+ case s: String => s
+ case l: List[_] => l.mkString(",")
+ case o if o != null => o.toString
+ case _ => entry.defaultValue
+ }
+ }
+
+}
+
+object AtlasClientConf {
+
+ private lazy val clientConf: AtlasClientConf = {
+ val conf = ApplicationProperties.get()
+ SparkContextHelper.globalSparkContext.getConf.getAllWithPrefix("spark.atlas.")
+ .foreach {
+ case (k, v) => conf.setProperty(s"atlas.$k", v)
+ }
Review Comment:
```suggestion
.foreach { case (k, v) => conf.setProperty(s"atlas.$k", v) }
```
Minor. Put in one line in a more compact code style.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Implements AtlasLineageDispatcher to send lineage to Apache Atlas
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191860128
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasEntityHelper.scala:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import scala.collection.JavaConverters._
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId, AtlasRelatedObjectId}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper
+
+object AtlasEntityHelper {
+
+ def processEntity(qe: QueryExecution, lineage: Lineage): AtlasEntity = {
+ val entity = new AtlasEntity(PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ val appName = SparkContextHelper.globalSparkContext.appName match {
+ case "Spark shell" => s"Spark Job + $appId"
+ case default => s"$default $appId"
+ }
+
+ entity.setAttribute("qualifiedName", appId)
+ entity.setAttribute("name", appName)
Review Comment:
We could use session confings for setting `appId` and `appName` in the followups, allowing user to custom the key details of the process entities.
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasEntityHelper.scala:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import scala.collection.JavaConverters._
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId, AtlasRelatedObjectId}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper
+
+object AtlasEntityHelper {
+
+ def processEntity(qe: QueryExecution, lineage: Lineage): AtlasEntity = {
+ val entity = new AtlasEntity(PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ val appName = SparkContextHelper.globalSparkContext.appName match {
+ case "Spark shell" => s"Spark Job + $appId"
+ case default => s"$default $appId"
+ }
+
+ entity.setAttribute("qualifiedName", appId)
+ entity.setAttribute("name", appName)
Review Comment:
We could use session configs for setting `appId` and `appName` in the followups, allowing user to custom the key details of the process entities.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] wForget commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191929561
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasEntityHelper.scala:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import scala.collection.JavaConverters._
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId, AtlasRelatedObjectId}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper
+
+object AtlasEntityHelper {
+
+ def processEntity(qe: QueryExecution, lineage: Lineage): AtlasEntity = {
+ val entity = new AtlasEntity(PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ val appName = SparkContextHelper.globalSparkContext.appName match {
+ case "Spark shell" => s"Spark Job + $appId"
+ case default => s"$default $appId"
+ }
+
+ entity.setAttribute("qualifiedName", appId)
+ entity.setAttribute("name", appName)
+ entity.setAttribute("currUser", SparkListenerHelper.currentUser)
+ entity.setAttribute("executionId", qe.id)
+ entity.setAttribute("details", qe.toString())
+ entity.setAttribute("sparkPlanDescription", qe.sparkPlan.toString())
+
+ // TODO add entity type instead of parsing from string
+ val inputs = lineage.inputTables.flatMap(tableObjectId(_)).map { objId =>
+ relatedObjectId(objId, RELATIONSHIP_DATASET_PROCESS_INPUTS)
+ }
+ val outputs = lineage.outputTables.flatMap(tableObjectId(_)).map { objId =>
+ relatedObjectId(objId, RELATIONSHIP_PROCESS_DATASET_OUTPUTS)
+ }
+
+ entity.setRelationshipAttribute("inputs", inputs.asJava)
+ entity.setRelationshipAttribute("outputs", outputs.asJava)
+
+ entity
+ }
+
+ def columnLineageEntities(processEntity: AtlasEntity, lineage: Lineage): Seq[AtlasEntity] = {
+ lineage.columnLineage.flatMap(columnLineage => {
+ val inputs = columnLineage.originalColumns.flatMap(columnObjectId(_)).map { objId =>
+ relatedObjectId(objId, RELATIONSHIP_DATASET_PROCESS_INPUTS)
+ }
+ val outputs = Option(columnLineage.column).flatMap(columnObjectId(_)).map { objId =>
+ relatedObjectId(objId, RELATIONSHIP_PROCESS_DATASET_OUTPUTS)
+ }.toSeq
+
+ if (!inputs.isEmpty && !outputs.isEmpty) {
+ val entity = new AtlasEntity(COLUMN_LINEAGE_TYPE)
+ val qualifiedName =
+ s"${processEntity.getAttribute("qualifiedName")}:${columnLineage.column}"
+ entity.setAttribute("qualifiedName", qualifiedName)
+ entity.setAttribute("name", qualifiedName)
+ entity.setRelationshipAttribute("inputs", inputs.asJava)
+ entity.setRelationshipAttribute("outputs", outputs.asJava)
+ entity.setRelationshipAttribute(
+ "process",
+ relatedObjectId(objectId(processEntity), RELATIONSHIP_SPARK_PROCESS_COLUMN_LINEAGE))
+ Some(entity)
+ } else {
+ None
+ }
+ })
+ }
+
+ def tableObjectId(tableName: String): Option[AtlasObjectId] = {
+ val dbTb = tableName.split('.')
+ if (dbTb.length == 2) {
Review Comment:
> Why Ignore ID generation for the table name with databas and catalog here?
I want to implement hive entity first, and then we can define a datasource type to replace the table string in order to support more entity types.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193272095
##########
extensions/spark/kyuubi-spark-lineage/src/test/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasLineageDispatcherSuite.scala:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import java.util
+
+import scala.collection.JavaConverters._
+import scala.collection.immutable.List
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId}
+import org.apache.commons.lang3.StringUtils
+import org.apache.spark.SparkConf
+import org.apache.spark.kyuubi.lineage.LineageConf.{DISPATCHERS, SKIP_PARSING_PERMANENT_VIEW_ENABLED}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.SparkListenerExtensionTest
+import org.scalatest.concurrent.PatienceConfiguration.Timeout
+import org.scalatest.time.SpanSugar._
+
+import org.apache.kyuubi.KyuubiFunSuite
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasEntityHelper.{COLUMN_LINEAGE_TYPE, PROCESS_TYPE}
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper.isSparkVersionAtMost
+
+class AtlasLineageDispatcherSuite extends KyuubiFunSuite with SparkListenerExtensionTest {
+ val catalogName =
+ if (isSparkVersionAtMost("3.1")) "org.apache.spark.sql.connector.InMemoryTableCatalog"
+ else "org.apache.spark.sql.connector.catalog.InMemoryTableCatalog"
+
+ override protected val catalogImpl: String = "hive"
+
+ override def sparkConf(): SparkConf = {
+ super.sparkConf()
+ .set("spark.sql.catalog.v2_catalog", catalogName)
+ .set(
+ "spark.sql.queryExecutionListeners",
+ "org.apache.kyuubi.plugin.lineage.SparkOperationLineageQueryExecutionListener")
+ .set(DISPATCHERS.key, "ATLAS")
+ .set(SKIP_PARSING_PERMANENT_VIEW_ENABLED.key, "true")
+ }
+
+ override def afterAll(): Unit = {
+ spark.stop()
+ super.afterAll()
+ }
+
+ test("altas lineage capture: insert into select sql") {
+ val mockAtlasClient = new MockAtlasClient()
+ AtlasClient.setClient(mockAtlasClient)
+
+ withTable("test_table0") { _ =>
+ spark.sql("create table test_table0(a string, b int, c int)")
+ spark.sql("create table test_table1(a string, d int)")
+ spark.sql("insert into test_table1 select a, b + c as d from test_table0").collect()
+ val expected = Lineage(
+ List("default.test_table0"),
+ List("default.test_table1"),
+ List(
+ ("default.test_table1.a", Set("default.test_table0.a")),
+ ("default.test_table1.d", Set("default.test_table0.b", "default.test_table0.c"))))
+ eventually(Timeout(5.seconds)) {
+ assert(mockAtlasClient.getEntities != null && mockAtlasClient.getEntities.nonEmpty)
+ }
+ checkAtlasProcessEntity(mockAtlasClient.getEntities.head, expected)
+ checkAtlasColumnLineageEntities(
+ mockAtlasClient.getEntities.head,
+ mockAtlasClient.getEntities.tail,
+ expected)
+ }
+
+ }
+
+ def checkAtlasProcessEntity(entity: AtlasEntity, expected: Lineage): Unit = {
+ assert(entity.getTypeName == PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ assert(entity.getAttribute("qualifiedName") == appId)
+ assert(entity.getAttribute("name")
+ == s"${SparkContextHelper.globalSparkContext.appName} $appId")
+ assert(StringUtils.isNotBlank(entity.getAttribute("currUser").asInstanceOf[String]))
+ assert(entity.getAttribute("executionId") != null)
+ assert(StringUtils.isNotBlank(entity.getAttribute("details").asInstanceOf[String]))
+ assert(StringUtils.isNotBlank(entity.getAttribute("sparkPlanDescription").asInstanceOf[String]))
+
+ val inputs = entity.getRelationshipAttribute("inputs")
+ .asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName)
+ val outputs = entity.getRelationshipAttribute("outputs")
+ .asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName)
+ assertResult(expected.inputTables.map(s => s"$s@$cluster"))(inputs)
+ assertResult(expected.outputTables.map(s => s"$s@$cluster"))(outputs)
+ }
+
+ def checkAtlasColumnLineageEntities(
+ processEntity: AtlasEntity,
+ entities: Seq[AtlasEntity],
+ expected: Lineage): Unit = {
+ assert(entities.size == expected.columnLineage.size)
+
+ entities.zip(expected.columnLineage).foreach {
+ case (entity, expectedLineage) =>
+ assert(entity.getTypeName == COLUMN_LINEAGE_TYPE)
+ val expectedQualifiedName =
+ s"${processEntity.getAttribute("qualifiedName")}:${expectedLineage.column}"
+ assert(entity.getAttribute("qualifiedName") == expectedQualifiedName)
+ assert(entity.getAttribute("name") == expectedQualifiedName)
+
+ val inputs = entity.getRelationshipAttribute("inputs")
+ .asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName)
+ assertResult(expectedLineage.originalColumns.map(s => s"$s@$cluster"))(inputs.toSet)
+
+ val outputs = entity.getRelationshipAttribute("outputs")
+ .asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName)
+ assert(outputs.size == 1)
+ assert(s"${expectedLineage.column}@$cluster" == outputs.head)
+
+ assert(getQualifiedName(entity.getRelationshipAttribute("process").asInstanceOf[
+ AtlasObjectId]) == processEntity.getAttribute("qualifiedName"))
+ }
+ }
+
+ // Configured in atlas-application.properties
+ val cluster = "test"
+
+ def getQualifiedName(objId: AtlasObjectId): String = {
+ objId.getUniqueAttributes.get("qualifiedName").asInstanceOf[String]
+ }
+
+ class MockAtlasClient() extends AtlasClient {
+ private var _entities: Seq[AtlasEntity] = null
Review Comment:
```suggestion
private var _entities: Seq[AtlasEntity] = _
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193265233
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasEntityHelper.scala:
##########
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import scala.collection.JavaConverters._
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId, AtlasRelatedObjectId}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper
+
+object AtlasEntityHelper {
+
+ def processEntity(qe: QueryExecution, lineage: Lineage): AtlasEntity = {
+ val entity = new AtlasEntity(PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ val appName = SparkContextHelper.globalSparkContext.appName match {
+ case "Spark shell" => s"Spark Job $appId"
+ case default => s"$default $appId"
+ }
+
+ entity.setAttribute("qualifiedName", appId)
+ entity.setAttribute("name", appName)
+ entity.setAttribute("currUser", SparkListenerHelper.currentUser)
+ SparkListenerHelper.sessionUser.foreach(entity.setAttribute("remoteUser", _))
+ entity.setAttribute("executionId", qe.id)
+ entity.setAttribute("details", qe.toString())
+ entity.setAttribute("sparkPlanDescription", qe.sparkPlan.toString())
+
+ // TODO add entity type instead of parsing from string
+ val inputs = lineage.inputTables.flatMap(tableObjectId(_)).map { objId =>
Review Comment:
```suggestion
val inputs = lineage.inputTables.flatMap(tableObjectId).map { objId =>
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] pan3793 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194792457
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasEntityHelper.scala:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import scala.collection.JavaConverters._
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId, AtlasRelatedObjectId}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper
+
+object AtlasEntityHelper {
+
+ def processEntity(qe: QueryExecution, lineage: Lineage): AtlasEntity = {
+ val entity = new AtlasEntity(PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ val appName = SparkContextHelper.globalSparkContext.appName match {
+ case "Spark shell" => s"Spark Job + $appId"
+ case default => s"$default $appId"
+ }
+
+ entity.setAttribute("qualifiedName", appId)
+ entity.setAttribute("name", appName)
+ entity.setAttribute("currUser", SparkListenerHelper.currentUser)
Review Comment:
let's add some comments to the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193452388
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -207,3 +208,24 @@ spark.sparkContext.addSparkListener(new SparkListener {
#### Get Lineage Events from Kyuubi EventHandler
When using the `KYUUBI_EVENT` dispatcher, the lineage events will be sent to the Kyuubi `EventBus`. Refer to [Kyuubi Event Handler](../../server/events) to handle kyuubi events.
+
+#### Ingest Lineage Entities to Apache Atlas
+
+The lineage entities can be ingested into [Apache Atlas](https://atlas.apache.org/) using the `ATLAS` dispatcher.
+
+Extra works:
+
++ The least transitive dependencies needed, which are under `./extensions/spark/kyuubi-spark-lineage/target/scala-${scala.binary.version}/jars`
++ Use `spark.files` to specify the `atlas-application.properties` configuration file for Atlas
+
+Atlas Client configurations (Configure in `atlas-application.properties` or passed in `spark.atlas.` prefix):
+
+| Name | Default Value | Description | Since |
+|------------------------------|------------------------|-------------------------------------------------------|-------|
+| atlas.rest.address | http://localhost:21000 | The rest endpoint url for the Atlas server | 1.8.0 |
+| atlas.client.type | rest | The client type (currently only supports rest) | 1.8.0 |
+| atlas.client.username | none | The client username | 1.8.0 |
+| atlas.client.password | none | The client password | 1.8.0 |
+| atlas.cluster.name | primary | The cluster name to use in qualifiedName of entities. | 1.8.0 |
+| atlas.column.lineage.enabled | true | Whether to ingest column lineages to Atlas. | 1.8.0 |
Review Comment:
```suggestion
| atlas.hook.spark.column.lineage.enabled | true | Whether to ingest column lineages to Atlas. | 1.8.0 |
```
This feature uses and serves as the implementation of Atlas's hook. I shall suggest use `atlas.hook.spark.` prefix for the custom configs here. The prefix`atlas.hook.` is aligned to the other hooks in Atlas, and the `spark` represents this is an Atlas hook implement for Spark.
WDYT? @wForget @pan3793 @iodone
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -207,3 +208,24 @@ spark.sparkContext.addSparkListener(new SparkListener {
#### Get Lineage Events from Kyuubi EventHandler
When using the `KYUUBI_EVENT` dispatcher, the lineage events will be sent to the Kyuubi `EventBus`. Refer to [Kyuubi Event Handler](../../server/events) to handle kyuubi events.
+
+#### Ingest Lineage Entities to Apache Atlas
+
+The lineage entities can be ingested into [Apache Atlas](https://atlas.apache.org/) using the `ATLAS` dispatcher.
+
+Extra works:
+
++ The least transitive dependencies needed, which are under `./extensions/spark/kyuubi-spark-lineage/target/scala-${scala.binary.version}/jars`
++ Use `spark.files` to specify the `atlas-application.properties` configuration file for Atlas
+
+Atlas Client configurations (Configure in `atlas-application.properties` or passed in `spark.atlas.` prefix):
+
+| Name | Default Value | Description | Since |
+|------------------------------|------------------------|-------------------------------------------------------|-------|
+| atlas.rest.address | http://localhost:21000 | The rest endpoint url for the Atlas server | 1.8.0 |
+| atlas.client.type | rest | The client type (currently only supports rest) | 1.8.0 |
+| atlas.client.username | none | The client username | 1.8.0 |
+| atlas.client.password | none | The client password | 1.8.0 |
+| atlas.cluster.name | primary | The cluster name to use in qualifiedName of entities. | 1.8.0 |
+| atlas.column.lineage.enabled | true | Whether to ingest column lineages to Atlas. | 1.8.0 |
Review Comment:
```suggestion
| atlas.hook.spark.column.lineage.enabled | true | Whether to ingest column lineages to Atlas. | 1.8.0 |
```
This feature uses and serves as the implementation of Atlas's hook. I shall suggest use `atlas.hook.spark.` prefix for the custom configs here. The prefix`atlas.hook.` is aligned to the other hooks in Atlas, and the `spark` represents this is an Atlas hook implemented for Spark.
WDYT? @wForget @pan3793 @iodone
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] pan3793 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194777327
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClient.scala:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import java.util.Locale
+
+import com.google.common.annotations.VisibleForTesting
+import org.apache.atlas.AtlasClientV2
+import org.apache.atlas.model.instance.AtlasEntity
+import org.apache.atlas.model.instance.AtlasEntity.AtlasEntitiesWithExtInfo
+import org.apache.commons.lang3.StringUtils
+import org.apache.hadoop.util.ShutdownHookManager
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf._
+
+trait AtlasClient {
+ def send(entities: Seq[AtlasEntity]): Unit
+
+ def close(): Unit
+}
+
+class AtlasRestClient(conf: AtlasClientConf) extends AtlasClient {
+
+ private val atlasClient: AtlasClientV2 = {
+ val serverUrl = conf.get(ATLAS_REST_ENDPOINT).split(",")
+ val username = conf.get(CLIENT_USERNAME)
+ val password = conf.get(CLIENT_PASSWORD)
+ if (StringUtils.isNoneBlank(username, password)) {
+ new AtlasClientV2(serverUrl, Array(username, password))
+ } else {
+ new AtlasClientV2(serverUrl: _*)
+ }
+ }
+
+ override def send(entities: Seq[AtlasEntity]): Unit = {
+ val entitiesWithExtInfo = new AtlasEntitiesWithExtInfo()
+ entities.foreach(entitiesWithExtInfo.addEntity)
+ atlasClient.createEntities(entitiesWithExtInfo)
+ }
+
+ override def close(): Unit = {
+ if (atlasClient != null) {
+ atlasClient.close()
+ }
+ }
+}
+
+object AtlasClient {
+
+ @volatile private var client: AtlasClient = _
Review Comment:
why not simply use lazy val?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194774969
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -207,3 +208,24 @@ spark.sparkContext.addSparkListener(new SparkListener {
#### Get Lineage Events from Kyuubi EventHandler
When using the `KYUUBI_EVENT` dispatcher, the lineage events will be sent to the Kyuubi `EventBus`. Refer to [Kyuubi Event Handler](../../server/events) to handle kyuubi events.
+
+#### Ingest Lineage Entities to Apache Atlas
+
+The lineage entities can be ingested into [Apache Atlas](https://atlas.apache.org/) using the `ATLAS` dispatcher.
+
+Extra works:
+
++ The least transitive dependencies needed, which are under `./extensions/spark/kyuubi-spark-lineage/target/scala-${scala.binary.version}/jars`
++ Use `spark.files` to specify the `atlas-application.properties` configuration file for Atlas
+
+Atlas Client configurations (Configure in `atlas-application.properties` or passed in `spark.atlas.` prefix):
Review Comment:
Not sure, I'm afraid not so?
1. `org.apache.atlas.ApplicationProperties#get` will load atlas-application.properties from class path, ( and `atlas.conf` path will be preferred if the system env `atlas.conf` set )
2. `AtlasClientConf#clientConf` in plugin loads spark's configs with `spark.atlas.` prefix and overrides
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] iodone commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "iodone (via GitHub)" <gi...@apache.org>.
iodone commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191872097
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasEntityHelper.scala:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import scala.collection.JavaConverters._
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId, AtlasRelatedObjectId}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper
+
+object AtlasEntityHelper {
+
+ def processEntity(qe: QueryExecution, lineage: Lineage): AtlasEntity = {
+ val entity = new AtlasEntity(PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ val appName = SparkContextHelper.globalSparkContext.appName match {
+ case "Spark shell" => s"Spark Job + $appId"
+ case default => s"$default $appId"
+ }
+
+ entity.setAttribute("qualifiedName", appId)
+ entity.setAttribute("name", appName)
+ entity.setAttribute("currUser", SparkListenerHelper.currentUser)
+ entity.setAttribute("executionId", qe.id)
+ entity.setAttribute("details", qe.toString())
+ entity.setAttribute("sparkPlanDescription", qe.sparkPlan.toString())
+
+ // TODO add entity type instead of parsing from string
+ val inputs = lineage.inputTables.flatMap(tableObjectId(_)).map { objId =>
+ relatedObjectId(objId, RELATIONSHIP_DATASET_PROCESS_INPUTS)
+ }
+ val outputs = lineage.outputTables.flatMap(tableObjectId(_)).map { objId =>
+ relatedObjectId(objId, RELATIONSHIP_PROCESS_DATASET_OUTPUTS)
+ }
+
+ entity.setRelationshipAttribute("inputs", inputs.asJava)
+ entity.setRelationshipAttribute("outputs", outputs.asJava)
+
+ entity
+ }
+
+ def columnLineageEntities(processEntity: AtlasEntity, lineage: Lineage): Seq[AtlasEntity] = {
+ lineage.columnLineage.flatMap(columnLineage => {
+ val inputs = columnLineage.originalColumns.flatMap(columnObjectId(_)).map { objId =>
+ relatedObjectId(objId, RELATIONSHIP_DATASET_PROCESS_INPUTS)
+ }
+ val outputs = Option(columnLineage.column).flatMap(columnObjectId(_)).map { objId =>
+ relatedObjectId(objId, RELATIONSHIP_PROCESS_DATASET_OUTPUTS)
+ }.toSeq
+
+ if (!inputs.isEmpty && !outputs.isEmpty) {
+ val entity = new AtlasEntity(COLUMN_LINEAGE_TYPE)
+ val qualifiedName =
+ s"${processEntity.getAttribute("qualifiedName")}:${columnLineage.column}"
+ entity.setAttribute("qualifiedName", qualifiedName)
+ entity.setAttribute("name", qualifiedName)
+ entity.setRelationshipAttribute("inputs", inputs.asJava)
+ entity.setRelationshipAttribute("outputs", outputs.asJava)
+ entity.setRelationshipAttribute(
+ "process",
+ relatedObjectId(objectId(processEntity), RELATIONSHIP_SPARK_PROCESS_COLUMN_LINEAGE))
+ Some(entity)
+ } else {
+ None
+ }
+ })
+ }
+
+ def tableObjectId(tableName: String): Option[AtlasObjectId] = {
+ val dbTb = tableName.split('.')
+ if (dbTb.length == 2) {
Review Comment:
Why Ignore ID generation for the table name with databas and catalog here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Implements AtlasLineageDispatcher to send lineage to Apache Atlas
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191828570
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasEntityHelper.scala:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import scala.collection.JavaConverters._
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId, AtlasRelatedObjectId}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper
+
+object AtlasEntityHelper {
+
+ def processEntity(qe: QueryExecution, lineage: Lineage): AtlasEntity = {
+ val entity = new AtlasEntity(PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ val appName = SparkContextHelper.globalSparkContext.appName match {
+ case "Spark shell" => s"Spark Job + $appId"
+ case default => default + s" $appId"
Review Comment:
```suggestion
case default => s"$default $appId"
```
Minor. Following the same template style as the previous one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] pan3793 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194763318
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -207,3 +208,24 @@ spark.sparkContext.addSparkListener(new SparkListener {
#### Get Lineage Events from Kyuubi EventHandler
When using the `KYUUBI_EVENT` dispatcher, the lineage events will be sent to the Kyuubi `EventBus`. Refer to [Kyuubi Event Handler](../../server/events) to handle kyuubi events.
+
+#### Ingest Lineage Entities to Apache Atlas
+
+The lineage entities can be ingested into [Apache Atlas](https://atlas.apache.org/) using the `ATLAS` dispatcher.
+
+Extra works:
+
++ The least transitive dependencies needed, which are under `./extensions/spark/kyuubi-spark-lineage/target/scala-${scala.binary.version}/jars`
++ Use `spark.files` to specify the `atlas-application.properties` configuration file for Atlas
+
+Atlas Client configurations (Configure in `atlas-application.properties` or passed in `spark.atlas.` prefix):
Review Comment:
those sentences are not consistent, BTW, does it work to put `atlas-application.properties` under `SPARK_CONF_DIR`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] pan3793 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1194771676
##########
extensions/spark/kyuubi-spark-lineage/pom.xml:
##########
@@ -88,7 +87,103 @@
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
- <scope>test</scope>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.jaxrs</groupId>
+ <artifactId>jackson-jaxrs-base</artifactId>
+ <version>${jackson.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
Review Comment:
I suppose it's still required, the exclusion aims to exclude from runtime scope, in this case, list them in the top-level w/ provided scope instead of exclusion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Implements AtlasLineageDispatcher to send lineage to Apache Atlas
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1190859851
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClientConf.scala:
##########
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import org.apache.atlas.ApplicationProperties
+import org.apache.commons.configuration.Configuration
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf.ConfigEntry
+
+class AtlasClientConf(configuration: Configuration) {
+
+ def get(entry: ConfigEntry): String = {
+ configuration.getProperty(entry.key) match {
+ case s: String => s
+ case l: List[_] => l.mkString(",")
+ case o if o != null => o.toString
+ case _ => entry.defaultValue
+ }
+ }
+
+}
+
+object AtlasClientConf {
+
+ private lazy val clientConf: AtlasClientConf = {
+ val conf = ApplicationProperties.get()
+ SparkContextHelper.globalSparkContext.getConf.getAllWithPrefix("spark.atlas.")
+ .foreach {
+ case (k, v) => conf.setProperty(s"atlas.$k", v)
+ }
+ new AtlasClientConf(conf)
+ }
+
+ def getConf(): AtlasClientConf = clientConf
+
+ case class ConfigEntry(key: String, defaultValue: String)
+
+ val ATLAS_REST_ENDPOINT = ConfigEntry("atlas.rest.address", "localhost:21000")
+
+ val CLIENT_TYPE = ConfigEntry("atlas.client.type", "rest")
+ val CLIENT_USERNAME = ConfigEntry("atlas.client.username", null)
+ val CLIENT_PASSWORD = ConfigEntry("atlas.client.password", null)
+
+ val CLUSTER_NAME = ConfigEntry("atlas.cluster.name", "primary")
Review Comment:
I'm not sure any default value should be provided to this config. The cluster name would be left null by default but to check it is required.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Implements AtlasLineageDispatcher to send lineage to Apache Atlas
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191816556
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClient.scala:
##########
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import java.util.Locale
+
+import com.google.common.annotations.VisibleForTesting
+import org.apache.atlas.AtlasClientV2
+import org.apache.atlas.model.instance.AtlasEntity
+import org.apache.atlas.model.instance.AtlasEntity.AtlasEntitiesWithExtInfo
+import org.apache.commons.lang3.StringUtils
+
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClientConf._
+
+trait AtlasClient {
+ def send(entities: Seq[AtlasEntity]): Unit
+}
+
+class AtlasRestClient(conf: AtlasClientConf) extends AtlasClient {
+
+ private val atlasClient: AtlasClientV2 = {
+ val serverUrl = conf.get(ATLAS_REST_ENDPOINT).split(",")
+ val username = conf.get(CLIENT_USERNAME)
+ val password = conf.get(CLIENT_PASSWORD)
+ if (StringUtils.isNotBlank(username) && StringUtils.isNotBlank(password)) {
Review Comment:
```suggestion
if (StringUtils.isNoneBlank(username, password)) {
```
minor suggestion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Implements AtlasLineageDispatcher to send lineage to Apache Atlas
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1191812980
##########
extensions/spark/kyuubi-spark-lineage/pom.xml:
##########
@@ -30,6 +30,10 @@
<name>Kyuubi Dev Spark Lineage Extension</name>
<url>https://kyuubi.apache.org/</url>
+ <properties>
+ <atlas.version>2.3.0</atlas.version>
Review Comment:
Move this property to parent pom?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193254931
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasLineageDispatcher.scala:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.plugin.lineage.{Lineage, LineageDispatcher}
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasEntityHelper.columnLineageEnable
+
+class AtlasLineageDispatcher extends LineageDispatcher with Logging {
+
+ override def send(qe: QueryExecution, lineageOpt: Option[Lineage]): Unit = {
+ try {
+ lineageOpt.filter(l => l.inputTables.nonEmpty || l.outputTables.nonEmpty).foreach(lineage => {
+ val processEntity = AtlasEntityHelper.processEntity(qe, lineage)
+ val columnLineageEntities = if (!lineage.columnLineage.isEmpty && columnLineageEnable) {
+ AtlasEntityHelper.columnLineageEntities(processEntity, lineage)
+ } else {
+ Seq()
+ }
Review Comment:
```suggestion
val columnLineageEntities = if (lineage.columnLineage.nonEmpty && columnLineageEnable) {
AtlasEntityHelper.columnLineageEntities(processEntity, lineage)
} else {
Seq.empty
}
```
Minor code improvement suggestion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193270672
##########
extensions/spark/kyuubi-spark-lineage/src/test/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasLineageDispatcherSuite.scala:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import java.util
+
+import scala.collection.JavaConverters._
+import scala.collection.immutable.List
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId}
+import org.apache.commons.lang3.StringUtils
+import org.apache.spark.SparkConf
+import org.apache.spark.kyuubi.lineage.LineageConf.{DISPATCHERS, SKIP_PARSING_PERMANENT_VIEW_ENABLED}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.SparkListenerExtensionTest
+import org.scalatest.concurrent.PatienceConfiguration.Timeout
+import org.scalatest.time.SpanSugar._
+
+import org.apache.kyuubi.KyuubiFunSuite
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasEntityHelper.{COLUMN_LINEAGE_TYPE, PROCESS_TYPE}
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper.isSparkVersionAtMost
+
+class AtlasLineageDispatcherSuite extends KyuubiFunSuite with SparkListenerExtensionTest {
+ val catalogName =
+ if (isSparkVersionAtMost("3.1")) "org.apache.spark.sql.connector.InMemoryTableCatalog"
+ else "org.apache.spark.sql.connector.catalog.InMemoryTableCatalog"
+
+ override protected val catalogImpl: String = "hive"
+
+ override def sparkConf(): SparkConf = {
+ super.sparkConf()
+ .set("spark.sql.catalog.v2_catalog", catalogName)
+ .set(
+ "spark.sql.queryExecutionListeners",
+ "org.apache.kyuubi.plugin.lineage.SparkOperationLineageQueryExecutionListener")
+ .set(DISPATCHERS.key, "ATLAS")
+ .set(SKIP_PARSING_PERMANENT_VIEW_ENABLED.key, "true")
+ }
+
+ override def afterAll(): Unit = {
+ spark.stop()
+ super.afterAll()
+ }
+
+ test("altas lineage capture: insert into select sql") {
+ val mockAtlasClient = new MockAtlasClient()
+ AtlasClient.setClient(mockAtlasClient)
+
+ withTable("test_table0") { _ =>
+ spark.sql("create table test_table0(a string, b int, c int)")
+ spark.sql("create table test_table1(a string, d int)")
+ spark.sql("insert into test_table1 select a, b + c as d from test_table0").collect()
+ val expected = Lineage(
+ List("default.test_table0"),
+ List("default.test_table1"),
+ List(
+ ("default.test_table1.a", Set("default.test_table0.a")),
+ ("default.test_table1.d", Set("default.test_table0.b", "default.test_table0.c"))))
+ eventually(Timeout(5.seconds)) {
+ assert(mockAtlasClient.getEntities != null && mockAtlasClient.getEntities.nonEmpty)
+ }
+ checkAtlasProcessEntity(mockAtlasClient.getEntities.head, expected)
+ checkAtlasColumnLineageEntities(
+ mockAtlasClient.getEntities.head,
+ mockAtlasClient.getEntities.tail,
+ expected)
+ }
+
+ }
+
+ def checkAtlasProcessEntity(entity: AtlasEntity, expected: Lineage): Unit = {
+ assert(entity.getTypeName == PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ assert(entity.getAttribute("qualifiedName") == appId)
+ assert(entity.getAttribute("name")
+ == s"${SparkContextHelper.globalSparkContext.appName} $appId")
+ assert(StringUtils.isNotBlank(entity.getAttribute("currUser").asInstanceOf[String]))
+ assert(entity.getAttribute("executionId") != null)
+ assert(StringUtils.isNotBlank(entity.getAttribute("details").asInstanceOf[String]))
+ assert(StringUtils.isNotBlank(entity.getAttribute("sparkPlanDescription").asInstanceOf[String]))
+
+ val inputs = entity.getRelationshipAttribute("inputs")
+ .asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName(_))
+ val outputs = entity.getRelationshipAttribute("outputs")
+ .asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName(_))
+ assertResult(expected.inputTables.map(s => s"$s@$cluster"))(inputs)
+ assertResult(expected.outputTables.map(s => s"$s@$cluster"))(outputs)
+ }
+
+ def checkAtlasColumnLineageEntities(
+ processEntity: AtlasEntity,
+ entities: Seq[AtlasEntity],
+ expected: Lineage): Unit = {
+ assert(entities.size == expected.columnLineage.size)
+
+ entities.zip(expected.columnLineage).foreach {
+ case (entity, expectedLineage) =>
+ assert(entity.getTypeName == COLUMN_LINEAGE_TYPE)
+ val expectedQualifiedName =
+ s"${processEntity.getAttribute("qualifiedName")}:${expectedLineage.column}"
+ assert(entity.getAttribute("qualifiedName") == expectedQualifiedName)
+ assert(entity.getAttribute("name") == expectedQualifiedName)
+
+ val inputs = entity.getRelationshipAttribute("inputs")
+ .asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName(_))
+ assertResult(expectedLineage.originalColumns.map(s => s"$s@$cluster"))(inputs.toSet)
+
+ val outputs = entity.getRelationshipAttribute("outputs")
+ .asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName(_))
Review Comment:
```suggestion
val outputs = entity.getRelationshipAttribute("outputs")
.asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName)
```
##########
extensions/spark/kyuubi-spark-lineage/src/test/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasLineageDispatcherSuite.scala:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import java.util
+
+import scala.collection.JavaConverters._
+import scala.collection.immutable.List
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId}
+import org.apache.commons.lang3.StringUtils
+import org.apache.spark.SparkConf
+import org.apache.spark.kyuubi.lineage.LineageConf.{DISPATCHERS, SKIP_PARSING_PERMANENT_VIEW_ENABLED}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.SparkListenerExtensionTest
+import org.scalatest.concurrent.PatienceConfiguration.Timeout
+import org.scalatest.time.SpanSugar._
+
+import org.apache.kyuubi.KyuubiFunSuite
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasEntityHelper.{COLUMN_LINEAGE_TYPE, PROCESS_TYPE}
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper.isSparkVersionAtMost
+
+class AtlasLineageDispatcherSuite extends KyuubiFunSuite with SparkListenerExtensionTest {
+ val catalogName =
+ if (isSparkVersionAtMost("3.1")) "org.apache.spark.sql.connector.InMemoryTableCatalog"
+ else "org.apache.spark.sql.connector.catalog.InMemoryTableCatalog"
+
+ override protected val catalogImpl: String = "hive"
+
+ override def sparkConf(): SparkConf = {
+ super.sparkConf()
+ .set("spark.sql.catalog.v2_catalog", catalogName)
+ .set(
+ "spark.sql.queryExecutionListeners",
+ "org.apache.kyuubi.plugin.lineage.SparkOperationLineageQueryExecutionListener")
+ .set(DISPATCHERS.key, "ATLAS")
+ .set(SKIP_PARSING_PERMANENT_VIEW_ENABLED.key, "true")
+ }
+
+ override def afterAll(): Unit = {
+ spark.stop()
+ super.afterAll()
+ }
+
+ test("altas lineage capture: insert into select sql") {
+ val mockAtlasClient = new MockAtlasClient()
+ AtlasClient.setClient(mockAtlasClient)
+
+ withTable("test_table0") { _ =>
+ spark.sql("create table test_table0(a string, b int, c int)")
+ spark.sql("create table test_table1(a string, d int)")
+ spark.sql("insert into test_table1 select a, b + c as d from test_table0").collect()
+ val expected = Lineage(
+ List("default.test_table0"),
+ List("default.test_table1"),
+ List(
+ ("default.test_table1.a", Set("default.test_table0.a")),
+ ("default.test_table1.d", Set("default.test_table0.b", "default.test_table0.c"))))
+ eventually(Timeout(5.seconds)) {
+ assert(mockAtlasClient.getEntities != null && mockAtlasClient.getEntities.nonEmpty)
+ }
+ checkAtlasProcessEntity(mockAtlasClient.getEntities.head, expected)
+ checkAtlasColumnLineageEntities(
+ mockAtlasClient.getEntities.head,
+ mockAtlasClient.getEntities.tail,
+ expected)
+ }
+
+ }
+
+ def checkAtlasProcessEntity(entity: AtlasEntity, expected: Lineage): Unit = {
+ assert(entity.getTypeName == PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ assert(entity.getAttribute("qualifiedName") == appId)
+ assert(entity.getAttribute("name")
+ == s"${SparkContextHelper.globalSparkContext.appName} $appId")
+ assert(StringUtils.isNotBlank(entity.getAttribute("currUser").asInstanceOf[String]))
+ assert(entity.getAttribute("executionId") != null)
+ assert(StringUtils.isNotBlank(entity.getAttribute("details").asInstanceOf[String]))
+ assert(StringUtils.isNotBlank(entity.getAttribute("sparkPlanDescription").asInstanceOf[String]))
+
+ val inputs = entity.getRelationshipAttribute("inputs")
+ .asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName(_))
+ val outputs = entity.getRelationshipAttribute("outputs")
+ .asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName(_))
Review Comment:
```suggestion
val inputs = entity.getRelationshipAttribute("inputs")
.asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName)
val outputs = entity.getRelationshipAttribute("outputs")
.asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName)
```
##########
extensions/spark/kyuubi-spark-lineage/src/test/scala/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasLineageDispatcherSuite.scala:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.dispatcher.atlas
+
+import java.util
+
+import scala.collection.JavaConverters._
+import scala.collection.immutable.List
+
+import org.apache.atlas.model.instance.{AtlasEntity, AtlasObjectId}
+import org.apache.commons.lang3.StringUtils
+import org.apache.spark.SparkConf
+import org.apache.spark.kyuubi.lineage.LineageConf.{DISPATCHERS, SKIP_PARSING_PERMANENT_VIEW_ENABLED}
+import org.apache.spark.kyuubi.lineage.SparkContextHelper
+import org.apache.spark.sql.SparkListenerExtensionTest
+import org.scalatest.concurrent.PatienceConfiguration.Timeout
+import org.scalatest.time.SpanSugar._
+
+import org.apache.kyuubi.KyuubiFunSuite
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasEntityHelper.{COLUMN_LINEAGE_TYPE, PROCESS_TYPE}
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper.isSparkVersionAtMost
+
+class AtlasLineageDispatcherSuite extends KyuubiFunSuite with SparkListenerExtensionTest {
+ val catalogName =
+ if (isSparkVersionAtMost("3.1")) "org.apache.spark.sql.connector.InMemoryTableCatalog"
+ else "org.apache.spark.sql.connector.catalog.InMemoryTableCatalog"
+
+ override protected val catalogImpl: String = "hive"
+
+ override def sparkConf(): SparkConf = {
+ super.sparkConf()
+ .set("spark.sql.catalog.v2_catalog", catalogName)
+ .set(
+ "spark.sql.queryExecutionListeners",
+ "org.apache.kyuubi.plugin.lineage.SparkOperationLineageQueryExecutionListener")
+ .set(DISPATCHERS.key, "ATLAS")
+ .set(SKIP_PARSING_PERMANENT_VIEW_ENABLED.key, "true")
+ }
+
+ override def afterAll(): Unit = {
+ spark.stop()
+ super.afterAll()
+ }
+
+ test("altas lineage capture: insert into select sql") {
+ val mockAtlasClient = new MockAtlasClient()
+ AtlasClient.setClient(mockAtlasClient)
+
+ withTable("test_table0") { _ =>
+ spark.sql("create table test_table0(a string, b int, c int)")
+ spark.sql("create table test_table1(a string, d int)")
+ spark.sql("insert into test_table1 select a, b + c as d from test_table0").collect()
+ val expected = Lineage(
+ List("default.test_table0"),
+ List("default.test_table1"),
+ List(
+ ("default.test_table1.a", Set("default.test_table0.a")),
+ ("default.test_table1.d", Set("default.test_table0.b", "default.test_table0.c"))))
+ eventually(Timeout(5.seconds)) {
+ assert(mockAtlasClient.getEntities != null && mockAtlasClient.getEntities.nonEmpty)
+ }
+ checkAtlasProcessEntity(mockAtlasClient.getEntities.head, expected)
+ checkAtlasColumnLineageEntities(
+ mockAtlasClient.getEntities.head,
+ mockAtlasClient.getEntities.tail,
+ expected)
+ }
+
+ }
+
+ def checkAtlasProcessEntity(entity: AtlasEntity, expected: Lineage): Unit = {
+ assert(entity.getTypeName == PROCESS_TYPE)
+
+ val appId = SparkContextHelper.globalSparkContext.applicationId
+ assert(entity.getAttribute("qualifiedName") == appId)
+ assert(entity.getAttribute("name")
+ == s"${SparkContextHelper.globalSparkContext.appName} $appId")
+ assert(StringUtils.isNotBlank(entity.getAttribute("currUser").asInstanceOf[String]))
+ assert(entity.getAttribute("executionId") != null)
+ assert(StringUtils.isNotBlank(entity.getAttribute("details").asInstanceOf[String]))
+ assert(StringUtils.isNotBlank(entity.getAttribute("sparkPlanDescription").asInstanceOf[String]))
+
+ val inputs = entity.getRelationshipAttribute("inputs")
+ .asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName(_))
+ val outputs = entity.getRelationshipAttribute("outputs")
+ .asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName(_))
+ assertResult(expected.inputTables.map(s => s"$s@$cluster"))(inputs)
+ assertResult(expected.outputTables.map(s => s"$s@$cluster"))(outputs)
+ }
+
+ def checkAtlasColumnLineageEntities(
+ processEntity: AtlasEntity,
+ entities: Seq[AtlasEntity],
+ expected: Lineage): Unit = {
+ assert(entities.size == expected.columnLineage.size)
+
+ entities.zip(expected.columnLineage).foreach {
+ case (entity, expectedLineage) =>
+ assert(entity.getTypeName == COLUMN_LINEAGE_TYPE)
+ val expectedQualifiedName =
+ s"${processEntity.getAttribute("qualifiedName")}:${expectedLineage.column}"
+ assert(entity.getAttribute("qualifiedName") == expectedQualifiedName)
+ assert(entity.getAttribute("name") == expectedQualifiedName)
+
+ val inputs = entity.getRelationshipAttribute("inputs")
+ .asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName(_))
Review Comment:
```suggestion
val inputs = entity.getRelationshipAttribute("inputs")
.asInstanceOf[util.Collection[AtlasObjectId]].asScala.map(getQualifiedName)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [kyuubi] bowenliang123 commented on a diff in pull request #4815: [KYUUBI #4814] Introduce Apache Atlas hook support in lineage plugin
Posted by "bowenliang123 (via GitHub)" <gi...@apache.org>.
bowenliang123 commented on code in PR #4815:
URL: https://github.com/apache/kyuubi/pull/4815#discussion_r1193293040
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -220,11 +220,12 @@ Extra works:
Atlas Client configurations (Configure in `atlas-application.properties` or passed in `spark.atlas.` prefix):
-| Name | Default Value | Description | Since |
-|-----------------------|-----------------|-------------------------------------------------------|-------|
-| atlas.rest.address | localhost:21000 | The rest endpoint url for the Atlas server | 1.8.0 |
-| atlas.client.type | rest | The client type (currently only supports rest) | 1.8.0 |
-| atlas.client.username | none | The client username | 1.8.0 |
-| atlas.client.password | none | The client password | 1.8.0 |
-| atlas.cluster.name | primary | The cluster name to use in qualifiedName of entities. | 1.8.0 |
+| Name | Default Value | Description | Since |
+|------------------------------|-----------------|-------------------------------------------------------|-------|
+| atlas.rest.address | localhost:21000 | The rest endpoint url for the Atlas server | 1.8.0 |
Review Comment:
Whether the protocol should be included as prefix in value?
I found all the default values WITH `http` prefix in Atlas's hook is as the following
`atlas.rest.address=http://localhost:31000`
(https://github.com/apache/atlas/blob/master/addons/hive-bridge/src/test/resources/atlas-application.properties#L20)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org