You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@atlas.apache.org by ch...@apache.org on 2025/04/16 07:32:34 UTC
(atlas) branch ATLAS-4984 created (now 72e787f91)
This is an automated email from the ASF dual-hosted git repository.
chaitalithombare pushed a change to branch ATLAS-4984
in repository https://gitbox.apache.org/repos/asf/atlas.git
at 72e787f91 ATLAS-4984: Option to ignore spark_process attributes details and sparkPlanDescription
This branch includes the following new commits:
new 72e787f91 ATLAS-4984: Option to ignore spark_process attributes details and sparkPlanDescription
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(atlas) 01/01: ATLAS-4984: Option to ignore spark_process attributes details and sparkPlanDescription
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chaitalithombare pushed a commit to branch ATLAS-4984
in repository https://gitbox.apache.org/repos/asf/atlas.git
commit 72e787f91eee8efb7c69da2ef0b79f154b7ca85f
Author: chaitalithombare <ch...@apache.org>
AuthorDate: Wed Apr 16 13:02:14 2025 +0530
ATLAS-4984: Option to ignore spark_process attributes details and sparkPlanDescription
---
.../notification/NotificationHookConsumer.java | 26 +++++-
.../preprocessor/EntityPreprocessor.java | 16 ++++
.../preprocessor/SparkPreprocessor.java | 38 ++++++++
.../preprocessor/SparkPreprocessorTest.java | 102 +++++++++++++++++++++
4 files changed, 180 insertions(+), 2 deletions(-)
diff --git a/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java b/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
index 49d378f85..a0d39ae90 100644
--- a/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
+++ b/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
@@ -156,6 +156,7 @@ public class NotificationHookConsumer implements Service, ActiveStateChangeHandl
public static final String CONSUMER_PREPROCESS_HIVE_TYPES_REMOVE_OWNEDREF_ATTRS = "atlas.notification.consumer.preprocess.hive_types.remove.ownedref.attrs";
public static final String CONSUMER_PREPROCESS_RDBMS_TYPES_REMOVE_OWNEDREF_ATTRS = "atlas.notification.consumer.preprocess.rdbms_types.remove.ownedref.attrs";
public static final String CONSUMER_PREPROCESS_S3_V2_DIRECTORY_PRUNE_OBJECT_PREFIX = "atlas.notification.consumer.preprocess.s3_v2_directory.prune.object_prefix";
+ public static final String CONSUMER_PREPROCESS_SPARK_PROCESS_ATTRIBUTES = "atlas.notification.consumer.preprocess.spark_process.attributes";
public static final String CONSUMER_AUTHORIZE_USING_MESSAGE_USER = "atlas.notification.authorize.using.message.user";
public static final String CONSUMER_AUTHORIZE_AUTHN_CACHE_TTL_SECONDS = "atlas.notification.authorize.authn.cache.ttl.seconds";
public static final int SERVER_READY_WAIT_TIME_MS = 1000;
@@ -202,6 +203,8 @@ public class NotificationHookConsumer implements Service, ActiveStateChangeHandl
private final boolean preprocessEnabled;
private final boolean createShellEntityForNonExistingReference;
private final boolean authorizeUsingMessageUser;
+ private final boolean sparkProcessAttributes;
+
private final Map<String, Authentication> authnCache;
private final NotificationInterface notificationInterface;
private final Configuration applicationProperties;
@@ -355,8 +358,8 @@ public class NotificationHookConsumer implements Service, ActiveStateChangeHandl
hiveTypesRemoveOwnedRefAttrs = applicationProperties.getBoolean(CONSUMER_PREPROCESS_HIVE_TYPES_REMOVE_OWNEDREF_ATTRS, true);
rdbmsTypesRemoveOwnedRefAttrs = applicationProperties.getBoolean(CONSUMER_PREPROCESS_RDBMS_TYPES_REMOVE_OWNEDREF_ATTRS, true);
s3V2DirectoryPruneObjectPrefix = applicationProperties.getBoolean(CONSUMER_PREPROCESS_S3_V2_DIRECTORY_PRUNE_OBJECT_PREFIX, true);
-
- preprocessEnabled = skipHiveColumnLineageHive20633 || updateHiveProcessNameWithQualifiedName || hiveTypesRemoveOwnedRefAttrs || rdbmsTypesRemoveOwnedRefAttrs || s3V2DirectoryPruneObjectPrefix || !hiveTablesToIgnore.isEmpty() || !hiveTablesToPrune.isEmpty() || !hiveDummyDatabasesToIgnore.isEmpty() || !hiveDummyTablesToIgnore.isEmpty() || !hiveTablePrefixesToIgnore.isEmpty();
+ sparkProcessAttributes = this.applicationProperties.getBoolean(CONSUMER_PREPROCESS_SPARK_PROCESS_ATTRIBUTES, false);
+ preprocessEnabled = skipHiveColumnLineageHive20633 || updateHiveProcessNameWithQualifiedName || hiveTypesRemoveOwnedRefAttrs || rdbmsTypesRemoveOwnedRefAttrs || s3V2DirectoryPruneObjectPrefix || !hiveTablesToIgnore.isEmpty() || !hiveTablesToPrune.isEmpty() || !hiveDummyDatabasesToIgnore.isEmpty() || !hiveDummyTablesToIgnore.isEmpty() || !hiveTablePrefixesToIgnore.isEmpty() || sparkProcessAttributes;
entityCorrelationManager = new EntityCorrelationManager(entityCorrelationStore);
LOG.info("{}={}", CONSUMER_SKIP_HIVE_COLUMN_LINEAGE_HIVE_20633, skipHiveColumnLineageHive20633);
@@ -591,6 +594,10 @@ public class NotificationHookConsumer implements Service, ActiveStateChangeHandl
pruneObjectPrefixForS3V2Directory(context);
}
+ if (sparkProcessAttributes) {
+ preprocessSparkProcessAttributes(context);
+ }
+
context.moveRegisteredReferredEntities();
if (context.isHivePreprocessEnabled() && CollectionUtils.isNotEmpty(context.getEntities()) && context.getEntities().size() > 1) {
@@ -700,6 +707,21 @@ public class NotificationHookConsumer implements Service, ActiveStateChangeHandl
}
}
+ private void preprocessSparkProcessAttributes(PreprocessorContext context) {
+ List<AtlasEntity> entities = context.getEntities();
+
+ if (entities != null) {
+ for (int i = 0; i < entities.size(); i++) {
+ AtlasEntity entity = entities.get(i);
+ EntityPreprocessor preprocessor = EntityPreprocessor.getSparkPreprocessor(entity.getTypeName());
+
+ if (preprocessor != null) {
+ preprocessor.preprocess(entity, context);
+ }
+ }
+ }
+ }
+
private void skipHiveColumnLineage(PreprocessorContext context) {
List<AtlasEntity> entities = context.getEntities();
diff --git a/webapp/src/main/java/org/apache/atlas/notification/preprocessor/EntityPreprocessor.java b/webapp/src/main/java/org/apache/atlas/notification/preprocessor/EntityPreprocessor.java
index 9cab8eda7..b801615f4 100644
--- a/webapp/src/main/java/org/apache/atlas/notification/preprocessor/EntityPreprocessor.java
+++ b/webapp/src/main/java/org/apache/atlas/notification/preprocessor/EntityPreprocessor.java
@@ -34,6 +34,7 @@ public abstract class EntityPreprocessor {
public static final String TYPE_HIVE_DB_DDL = "hive_db_ddl";
public static final String TYPE_HIVE_TABLE_DDL = "hive_table_ddl";
public static final String TYPE_HIVE_TABLE = "hive_table";
+ public static final String TYPE_SPARK_PROCESS = "spark_process";
public static final String TYPE_RDBMS_INSTANCE = "rdbms_instance";
public static final String TYPE_RDBMS_DB = "rdbms_db";
public static final String TYPE_RDBMS_TABLE = "rdbms_table";
@@ -42,6 +43,8 @@ public abstract class EntityPreprocessor {
public static final String TYPE_RDBMS_FOREIGN_KEY = "rdbms_foreign_key";
public static final String ATTRIBUTE_COLUMNS = "columns";
+ public static final String ATTRIBUTE_DETAILS = "details";
+ public static final String ATTRIBUTE_SPARKPLANDESCRIPTION = "sparkPlanDescription";
public static final String ATTRIBUTE_INPUTS = "inputs";
public static final String ATTRIBUTE_OUTPUTS = "outputs";
public static final String ATTRIBUTE_PARTITION_KEYS = "partitionKeys";
@@ -66,6 +69,7 @@ public abstract class EntityPreprocessor {
private static final Map<String, EntityPreprocessor> HIVE_PREPROCESSOR_MAP = new HashMap<>();
private static final Map<String, EntityPreprocessor> RDBMS_PREPROCESSOR_MAP = new HashMap<>();
private static final Map<String, EntityPreprocessor> AWS_S3_V2_PREPROCESSOR_MAP = new HashMap<>();
+ private static final Map<String, EntityPreprocessor> SPARK_PREPROCESSOR_MAP = new HashMap<>();
private final String typeName;
@@ -85,6 +89,10 @@ public abstract class EntityPreprocessor {
return typeName != null ? AWS_S3_V2_PREPROCESSOR_MAP.get(typeName) : null;
}
+ public static EntityPreprocessor getSparkPreprocessor(String typeName) {
+ return typeName != null ? SPARK_PREPROCESSOR_MAP.get(typeName) : null;
+ }
+
public static String getQualifiedName(AtlasEntity entity) {
Object obj = entity != null ? entity.getAttribute(ATTRIBUTE_QUALIFIED_NAME) : null;
@@ -175,6 +183,10 @@ public abstract class EntityPreprocessor {
new AWSS3V2Preprocessor.AWSS3V2DirectoryPreprocessor()
};
+ EntityPreprocessor[] sparkPreprocessors = new EntityPreprocessor[]{
+ new SparkPreprocessor.SparkProcessPreprocessor()
+ };
+
for (EntityPreprocessor preprocessor : hivePreprocessors) {
HIVE_PREPROCESSOR_MAP.put(preprocessor.getTypeName(), preprocessor);
}
@@ -186,5 +198,9 @@ public abstract class EntityPreprocessor {
for (EntityPreprocessor preprocessor : s3V2Preprocessors) {
AWS_S3_V2_PREPROCESSOR_MAP.put(preprocessor.getTypeName(), preprocessor);
}
+
+ for (EntityPreprocessor preprocessor : sparkPreprocessors) {
+ SPARK_PREPROCESSOR_MAP.put(preprocessor.getTypeName(), preprocessor);
+ }
}
}
diff --git a/webapp/src/main/java/org/apache/atlas/notification/preprocessor/SparkPreprocessor.java b/webapp/src/main/java/org/apache/atlas/notification/preprocessor/SparkPreprocessor.java
new file mode 100644
index 000000000..266ce3790
--- /dev/null
+++ b/webapp/src/main/java/org/apache/atlas/notification/preprocessor/SparkPreprocessor.java
@@ -0,0 +1,38 @@
+/**
+ * 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.atlas.notification.preprocessor;
+
+import org.apache.atlas.model.instance.AtlasEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SparkPreprocessor {
+ private static final Logger LOG = LoggerFactory.getLogger(SparkPreprocessor.class);
+ static class SparkProcessPreprocessor extends EntityPreprocessor {
+ public SparkProcessPreprocessor() {
+ super(TYPE_SPARK_PROCESS);
+ }
+
+ @Override
+ public void preprocess(AtlasEntity entity, PreprocessorContext context) {
+ entity.removeAttribute(ATTRIBUTE_DETAILS);
+ entity.removeAttribute(ATTRIBUTE_SPARKPLANDESCRIPTION);
+ }
+ }
+}
diff --git a/webapp/src/test/java/org/apache/atlas/notification/preprocessor/SparkPreprocessorTest.java b/webapp/src/test/java/org/apache/atlas/notification/preprocessor/SparkPreprocessorTest.java
new file mode 100644
index 000000000..ec8927364
--- /dev/null
+++ b/webapp/src/test/java/org/apache/atlas/notification/preprocessor/SparkPreprocessorTest.java
@@ -0,0 +1,102 @@
+/**
+ * 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.atlas.notification.preprocessor;
+
+import org.apache.atlas.kafka.AtlasKafkaMessage;
+import org.apache.atlas.kafka.KafkaNotification;
+import org.apache.atlas.model.instance.AtlasEntity;
+import org.apache.atlas.model.notification.HookNotification;
+import org.apache.atlas.notification.hook.HookMessageDeserializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.Test;
+
+import java.util.Map;
+import java.util.HashMap;
+import java.util.List;
+import java.util.ArrayList;
+
+import java.util.regex.Pattern;
+
+import static org.testng.Assert.assertEquals;
+
+public class SparkPreprocessorTest {
+ private static final Logger LOG = LoggerFactory.getLogger(SparkPreprocessorTest.class);
+ private final HookMessageDeserializer deserializer = new HookMessageDeserializer();
+ public static final String TYPE_SPARK_PROCESS = "spark_process";
+ public static final String ATTRIBUTE_DETAILS = "details";
+ public static final String ATTRIBUTE_SPARKPLANDESCRIPTION = "sparkPlanDescription";
+ public static final String ATTRIBUTE_QUALIFIED_NAME = "qualifiedName";
+ public static final String ATTRIBUTE_NAME = "name";
+ public static final String ATTRIBUTE_ISINCOMPLETE = "isIncomplete";
+ public static final String ATTRIBUTE_REMOTEUSER = "remoteUser";
+ public static final String ATTRIBUTE_EXECUTIONID = "executionId";
+ public static final String ATTRIBUTE_QUERYTEXT = "queryText";
+ public static final String ATTRIBUTE_CURRUSER = "currUser";
+ public static final String ATTRIBUTE_GUID = "guid";
+ private static final List<String> EMPTY_STR_LIST = new ArrayList<>();
+ private static final List<Pattern> EMPTY_PATTERN_LIST = new ArrayList<>();
+
+ private void getPreprocessorContext(AtlasEntity entity) {
+ EntityPreprocessor preprocessor = EntityPreprocessor.getSparkPreprocessor(entity.getTypeName());
+ HookNotification hookNotification = new HookNotification.EntityCreateRequestV2("test", new AtlasEntity.AtlasEntitiesWithExtInfo(entity));
+
+ AtlasKafkaMessage<HookNotification> kafkaMsg = new AtlasKafkaMessage(hookNotification, -1, KafkaNotification.ATLAS_HOOK_TOPIC, -1);
+
+ PreprocessorContext context = new PreprocessorContext(kafkaMsg, null, EMPTY_PATTERN_LIST, EMPTY_PATTERN_LIST, null,
+ EMPTY_STR_LIST, EMPTY_STR_LIST, EMPTY_STR_LIST, false,
+ false, true, false, null);
+ if (preprocessor != null) {
+ preprocessor.preprocess(entity, context);
+ }
+ }
+
+ public Object[][] provideSparkProcessData() {
+ Map<String, Object> attributes = new HashMap<>();
+ attributes.put(ATTRIBUTE_NAME, "execution-1");
+ attributes.put(ATTRIBUTE_QUALIFIED_NAME, "application_1740993925593_0006-execution-1sparkTab1");
+ attributes.put(ATTRIBUTE_DETAILS, "== Parsed Logical Plan ==\\nCreateHiveTableAsSelectCommand ...");
+ attributes.put(ATTRIBUTE_SPARKPLANDESCRIPTION, "Execute CreateHiveTableAsSelectCommand ...");
+ attributes.put(ATTRIBUTE_GUID, "-32055574130361399");
+ attributes.put(ATTRIBUTE_ISINCOMPLETE, "false");
+ attributes.put(ATTRIBUTE_REMOTEUSER, "spark");
+ attributes.put(ATTRIBUTE_EXECUTIONID, 1);
+ attributes.put(ATTRIBUTE_QUERYTEXT, null);
+ attributes.put(ATTRIBUTE_CURRUSER, "spark");
+
+ return new Object[][] { { attributes } };
+ }
+
+ @Test
+ public void replaceAttributesInSparkProcess() {
+ Object[][] testData = provideSparkProcessData();
+
+ for (Object[] data : testData) {
+ Map<String, Object> attributes = (Map<String, Object>) data[0]; // Extract attributes
+
+ AtlasEntity atlasEntity = new AtlasEntity();
+ atlasEntity.setTypeName(TYPE_SPARK_PROCESS);
+ attributes.forEach(atlasEntity::setAttribute);
+ getPreprocessorContext(atlasEntity);
+
+ assertEquals(atlasEntity.getAttribute(ATTRIBUTE_DETAILS), null);
+ assertEquals(atlasEntity.getAttribute(ATTRIBUTE_SPARKPLANDESCRIPTION), null);
+ }
+ }
+}