ATLAS-837 Enhance Sqoop addon to handle export operation (venkatnrangan via shwethags)

addons/sqoop-bridge/src/main/java/org/apache/atlas/sqoop/hook/SqoopHook.java

@@ -109,8 +109,13 @@ public class SqoopHook extends SqoopJobDataPublisher {
         procRef.set(SqoopDataModelGenerator.NAME, sqoopProcessName);
         procRef.set(AtlasClient.REFERENCEABLE_ATTRIBUTE_NAME, sqoopProcessName);
         procRef.set(SqoopDataModelGenerator.OPERATION, data.getOperation());
+        if (isImportOperation(data)) {
             procRef.set(SqoopDataModelGenerator.INPUTS, dbStoreRef);
             procRef.set(SqoopDataModelGenerator.OUTPUTS, hiveTableRef);
+        } else {
+            procRef.set(SqoopDataModelGenerator.INPUTS, hiveTableRef);
+            procRef.set(SqoopDataModelGenerator.OUTPUTS, dbStoreRef);
+        }
         procRef.set(SqoopDataModelGenerator.USER, data.getUser());
         procRef.set(SqoopDataModelGenerator.START_TIME, new Date(data.getStartTime()));
         procRef.set(SqoopDataModelGenerator.END_TIME, new Date(data.getEndTime()));
@@ -126,15 +131,16 @@ public class SqoopHook extends SqoopJobDataPublisher {
     }
 
     static String getSqoopProcessName(Data data, String clusterName) {
-        StringBuilder name = new StringBuilder(String.format("sqoop import --connect %s", data.getUrl()));
+        StringBuilder name = new StringBuilder(String.format("sqoop %s --connect %s", data.getOperation(),
+                data.getUrl()));
         if (StringUtils.isNotEmpty(data.getStoreTable())) {
             name.append(" --table ").append(data.getStoreTable());
         }
         if (StringUtils.isNotEmpty(data.getStoreQuery())) {
             name.append(" --query ").append(data.getStoreQuery());
         }
-        name.append(String.format(" --hive-import --hive-database %s --hive-table %s --hive-cluster %s",
-                data.getHiveDB().toLowerCase(), data.getHiveTable().toLowerCase(), clusterName));
+        name.append(String.format(" --hive-%s --hive-database %s --hive-table %s --hive-cluster %s",
+                data.getOperation(), data.getHiveDB().toLowerCase(), data.getHiveTable().toLowerCase(), clusterName));
         return name.toString();
     }
 
@@ -149,6 +155,10 @@ public class SqoopHook extends SqoopJobDataPublisher {
         return name.toString();
     }
 
+    static boolean isImportOperation(SqoopJobDataPublisher.Data data) {
+        return data.getOperation().toLowerCase().equals("import");
+    }
+
     @Override
     public void publish(SqoopJobDataPublisher.Data data) throws Exception {
         Configuration atlasProperties = ApplicationProperties.get();

addons/sqoop-bridge/src/test/java/org/apache/atlas/sqoop/hook/SqoopHookIT.java

@@ -88,6 +88,21 @@ public class SqoopHookIT {
         assertHiveTableIsRegistered(DEFAULT_DB, "hiveTable");
     }
 
+    @Test
+    public void testSqoopExport() throws Exception {
+        SqoopJobDataPublisher.Data d = new SqoopJobDataPublisher.Data("export", "jdbc:mysql:///localhost/db",
+                "mysqluser", "mysql", "myTable", null, "default", "hiveTable", new Properties(),
+                System.currentTimeMillis() - 100, System.currentTimeMillis());
+        SqoopHook hook = new SqoopHook();
+        hook.publish(d);
+        Thread.sleep(1000);
+        String storeName  = SqoopHook.getSqoopDBStoreName(d);
+        assertDBStoreIsRegistered(storeName);
+        String name = SqoopHook.getSqoopProcessName(d, CLUSTER_NAME);
+        assertSqoopProcessIsRegistered(name);
+        assertHiveTableIsRegistered(DEFAULT_DB, "hiveTable");
+    }
+
     private String assertDBStoreIsRegistered(String storeName) throws Exception {
         LOG.debug("Searching for db store {}",  storeName);
         String query = String.format(

release-log.txt

@@ -22,6 +22,7 @@ ATLAS-409 Atlas will not import avro tables with schema read from a file (dosset
 ATLAS-379 Create sqoop and falcon metadata addons (venkatnrangan,bvellanki,sowmyaramesh via shwethags)
 
 ALL CHANGES:
+ATLAS-837 Enhance Sqoop addon to handle export operation (venkatnrangan via shwethags)
 ATLAS-869 Make LDAP/AD properties to be configurable. (nixonrodrigues via yhemanth)
 ATLAS-730 Change titan hbase table name (svimal2106 via sumasai )
 ATLAS-871 Make audit repository implementation configurable (jnhagelb via shwethags)