General 

To support use cases of migrating files from OnPrem to Google cloud we there is a need to for comphrensive support for Filesfile handling capabilities. This includes FileList, FileCompression, FileDecompression, FileEncryption , FileDecryption etc.  Right now the Plugins thats supported on CDAP do not support Fileconstructs but rather at the Record level There are few file level plugins available in CDAP like FileMove, FileDelete and this needs to be expanded. 

UseCase 1

Proposed Design 

Image RemovedImage Added

1. FileList Plugin - BatchSource Plugin  - Implement a new FileList plugin ( Batchsource plugin) with similar capability of the current FileSource plugin but instead of actually reading the file contents  it would just pass the filenames with full URI to be used for processing the following actions in pipeline. FileCompress
2. FileCompressEncrypt Sink Plugin - Transform  SparkSink Plugin -  Implement   Implement a Compression plugin similar to the Field Compression Plugin which accepts an input file URI , reads the file compress this and stored it temporarily on teh same node and spits out the compressed file URI location that can be used by the next processing action.   Invoke the current Google Cloud Storage plugin to persist the file to Cloud storagenew plugin of type SparkSink that will read the filename's from the FileList plugin , will compress the file using gzip / snappy then encrypt the file using PGP Public key and persist to Google Cloud storage or HDFS or FileSystem. 

FileList Plugin 

This is a Batchsource plugin similar to current FileSource Plugin but only list the filenames with full URI and not actually read the contents of the file. 

Plugin

...

Properties

Implementation Logic 

Base the code off FileSource Plugin  . Override the PathTrackingInputFormat  ( I have checked the PathTrackingBlobInputFormat, probably I can take that and just overwrite without a deligate and use that ).  

Use PathTrackingBlobInputFormat 

@Override
public StructuredRecord.Builder getCurrentValue() {
  String fieldName = schema.getFields().iterator().next().getName();
  return StructuredRecord.builder(schema).set(fieldName, val  Instead of actual record just pass in file path with name of the file ?);
}

...

Queries 

• If SFTP or FTP needs to be supported then its not clear how the credential information can be shared to the next step in the process. 

FileCompressEncrypt Plugin

This plugin will take input file name thats passed from the FileList Plugin, Get the fileInputStream using the URI and then using Gzip or Snappy libs compress the file and store them locally on the node, encrypt the file using PGP public key and persist the file. 

Plugin Properties

 Implement a new Transform Plugin for Compression,  use the exiting FieldCompression Plugin as starting point with following modification  

...

Open the file , get the InputFile Stream and pass this to the GZIP compression API's and store the file locally on the node. 

Queries 

...

Queries 

1. What is the best approach to track processed files so they are not processed again.  Proposing moving the input files after successful processing to a different directory so they dont get processed again in the next run. 

Usecase 2

Image Added

FileDecompressDecrypt Plugin

The plugin support decrypting  files using PGP Public key and decompress file. 

Plugin Properties