[jira] [Commented] (SQOOP-1803) JobManager and Execution Engine changes: Support for a injecting and pulling out configs and job output in connectors

["Jarek Jarcec Cecho (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/SQOOP-1803?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14345504#comment-14345504 ] 

Jarek Jarcec Cecho commented on SQOOP-1803:
-------------------------------------------

I was thinking about this one a bit myself. I have couple of thoughts of getting data back from the execution engine and I'm wondering what others thinks. Please don't hesitate and chime in if I missed any approach.

1) DistributedCache

In addition to [~gwenshap] comments about supportability (and/or debuggability) of {{DistributedCache}}, to my best knowledge it can be only used to distribute data from the launcher (Sqoop Server) to children mapreduce tasks. I do not believe that it can be used to the other way around to get files or data from individual tasks back to the launcher. Looking at the [latest javadocs|https://hadoop.apache.org/docs/current/api/org/apache/hadoop/filecache/DistributedCache.html] it seems still valid as the documentation contains note about immutability of the cache when the job is submitted:

{quote}
DistributedCache tracks modification timestamps of the cache files. Clearly the cache files should not be modified by the application or externally while the job is executing.
{quote}

*Summary:* I believe that this solution is disqualified for the retrieving data back from execution engine.

2) Counters

[Counters|https://hadoop.apache.org/docs/current/api/org/apache/hadoop/mapred/Counters.html] are nice technique how to get insight into what is the mapreduce job doing. Multiple mappers can be updating the counters in parallel and it's mapreduce responsibility to ensure that counters from all children tasks are sum-upped correctly. The limitation of this solution is that counters can be only {{long}} based (e.g. no {{String}}, {{Date}}, ...). Also the counters are cumulative in nature, so it might be a bit difficult to retrieve discrete values - we would need to ensure that only certain mappers/reducers will update given counters whereas others won't or we would need to figure out a way how to decode one single value when multiple mapper/reducers will update one counter at the same time.

*Summary:* Whereas it's not impossible to use the counters to retrieve data from execution engine, it seems that this solution will impose limitations and will be "difficult" to implement and maintain.

3) HDFS Files

Looking into how others are solving this problem, [Oozie|http://oozie.apache.org] launcher tasks (=one map mapreduce jobs) are generating files on HDFS in predefined directory from where the Oozie server will pick them up to read any arbitrary values. This is neat solution as it allows us to retrieve any value of any type from any part of the workflow (all processes can create their own files if needed). The downside is that we would need to agree on certain location where Server and the mapreduce job will be exchanging files - this directory must exists and must be accessible by both Sqoop (running under system user) and the mapreduce job itself (most likely running as end user). I believe that HDFS ACLs can be easily used to accomplish this task.

We would need to be careful here with edge conditions - we would need to make sure that we're cleaning up old and unused files (job failures, ...) and that we are not leaking any sensitive information to the HDFS.

*Summary:* Possible solution that will support all our use cases, but will be a bit harder to implement.

4) Server side exchange only

I was also looking into how things work currently in the server and I've realized something that made me thing about this proposal. Back when we were defining the workflow, the intention was that only {{Initializer}} is allowed to generate state whereas all other parts of the workflow {{Partitioner}}, {{Extractor}}, {{Loader}} and {{Destroyer}} should not generate any state and only reuse the one that was pre-prepared in initializer. The reason for that is that {{Initializer}} is run only once whereas all other parts of the workflow are run in parallel and/or not running on Sqoop server itself, hence by allowing state to be generated only in {{Initializer}} we don't have to deal with synchronizing the parallel pieces or deal with limitations in various execution engines. The intention is persisted in the API when {{Initializer}} is given {{MutableContext}} where connector developer can set any properties that will be shared with rest of the workflow (~ the state) and when all other parts are given only {{ImmutableContext}} that doesn't allow any changes to the shared properties. I have to say that we have small exception in the code base, because {{Partitioner}} class is generating {{Partition}} objects that can carry some context as well. However as the {{Partition}} objects are not available in {{Destroyer}}, connector developer still needs to persist state that is required through entire workflow inside the {{Initializer}}.

Having said that, another option seems to be to simply not retrieve anything from the execution engine and let connector update the configuration objects based on info that the connector generated in {{Initializer}} - assuming that the job finished correctly. Looking at current connectors this should work well, as we need to update and persist state that is 'locked' at the {{Initializer}} stage. For database-base connectors the "max value" should be determined in initializer (it's currently not though) and the same for Kafka and other connectors. The beauty of this approach is that it's simple to implement and can actually be easily extended in the future to include data coming from execution engine shell there be a need for it (for the approach 3) for example).

> JobManager and Execution Engine changes: Support for a injecting and pulling out configs and job output in connectors 
> ----------------------------------------------------------------------------------------------------------------------
>
>                 Key: SQOOP-1803
>                 URL: https://issues.apache.org/jira/browse/SQOOP-1803
>             Project: Sqoop
>          Issue Type: Sub-task
>            Reporter: Veena Basavaraj
>            Assignee: Veena Basavaraj
>             Fix For: 1.99.6
>
>
> The details are in the design wiki, as the implementation happens more discussions can happen here.
> https://cwiki.apache.org/confluence/display/SQOOP/Delta+Fetch+And+Merge+Design#DeltaFetchAndMergeDesign-Howtogetoutputfromconnectortosqoop?
> The goal is to dynamically inject a IncrementalConfig instance into the FromJobConfiguration. The current MFromConfig and MToConfig can already hold a list of configs, and a strong sentiment was expressed to keep it as a list, why not for the first time actually make use of it and group the incremental related configs in one config object
> This task will prepare the FromJobConfiguration from the job config data, ExtractorContext with the relevant values from the prev job run 
> This task will prepare the ToJobConfiguration from the job config data, LoaderContext with the relevant values from the prev job run if any
> We will use DistributedCache to get State information from the Extractor and Loader out and finally persist it into the sqoop repository depending on SQOOP-1804 once the outputcommitter commit is called



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)