<commit_before>/**
 * Copyright 2016 Otto (GmbH & Co KG)
 *
 * Licensed 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.schedoscope.export.ftp;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.util.ToolRunner;
import org.apache.hive.hcatalog.mapreduce.HCatInputFormat;
import org.kohsuke.args4j.CmdLineException;
import org.kohsuke.args4j.CmdLineParser;
import org.kohsuke.args4j.Option;
import org.schedoscope.export.BaseExportJob;
import org.schedoscope.export.ftp.outputformat.CSVOutputFormat;
import org.schedoscope.export.ftp.upload.FileCompressionCodec;
import org.schedoscope.export.writables.TextPairArrayWritable;

public class FtpExportJob extends BaseExportJob {

	private static final Log LOG = LogFactory.getLog(FtpExportJob.class);

	@Option(name = "-k", usage = "ssh private key file")
	private String keyFile = "~/.ssh/id_rsa";

	@Option(name = "-u", usage = "the (s)ftp user", required = true)
	private String ftpUser;

	@Option(name = "-w", usage = "the ftp password / sftp passphrase")
	private String ftpPass;

	@Option(name = "-j", usage = "the (s)ftp endpoint, e.g. ftp://ftp.example.com:21/path/to/", required = true)
	private String ftpEndpoint;

	@Option(name = "-x", usage = "passive mode, only for ftp connections, defaults to 'true'")
	private boolean passiveMode = true;

	@Option(name = "-z", usage = "user dir is root, home dir on remote end is (s)ftp root dir defaults to 'true'")
	private boolean userIsRoot = true;

	@Option(name = "-g", usage = "clean up hdfs dir after export, defaults to 'true'")
	private boolean cleanHdfsDir = true;

	@Option(name = "-c", usage = "compression codec, either 'none', 'gzip' or 'bzip2', defaults to 'gzip'")
	private FileCompressionCodec codec = FileCompressionCodec.gzip;

	@Override
	public int run(String[] args) throws Exception {

		CmdLineParser cmd = new CmdLineParser(this);

		try {
			cmd.parseArgument(args);
		} catch (CmdLineException e) {
			System.err.println(e.getMessage());
			cmd.printUsage(System.err);
			throw e;
		}

		Job job = configure();
		return job.waitForCompletion(true) ? 0 : 1;
	}

	private Job configure() throws Exception {

		Configuration conf = getConfiguration();
		conf = configureHiveMetaStore(conf);
		conf = configureKerberos(conf);
		conf = configureAnonFields(conf);

		Job job = Job.getInstance(conf, "FtpExport: " + inputDatabase + "." + inputTable);<commit_msg>add configure function to trigger job from external modules<commit_after>/**
 * Copyright 2016 Otto (GmbH & Co KG)
 *
 * Licensed 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.schedoscope.export.ftp;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.util.ToolRunner;
import org.apache.hive.hcatalog.mapreduce.HCatInputFormat;
import org.kohsuke.args4j.CmdLineException;
import org.kohsuke.args4j.CmdLineParser;
import org.kohsuke.args4j.Option;
import org.schedoscope.export.BaseExportJob;
import org.schedoscope.export.ftp.outputformat.CSVOutputFormat;
import org.schedoscope.export.ftp.upload.FileCompressionCodec;
import org.schedoscope.export.writables.TextPairArrayWritable;

public class FtpExportJob extends BaseExportJob {

	private static final Log LOG = LogFactory.getLog(FtpExportJob.class);

	@Option(name = "-k", usage = "ssh private key file")
	private String keyFile = "~/.ssh/id_rsa";

	@Option(name = "-u", usage = "the (s)ftp user", required = true)
	private String ftpUser;

	@Option(name = "-w", usage = "the ftp password / sftp passphrase")
	private String ftpPass;

	@Option(name = "-j", usage = "the (s)ftp endpoint, e.g. ftp://ftp.example.com:21/path/to/", required = true)
	private String ftpEndpoint;

	@Option(name = "-x", usage = "passive mode, only for ftp connections, defaults to 'true'")
	private boolean passiveMode = true;

	@Option(name = "-z", usage = "user dir is root, home dir on remote end is (s)ftp root dir defaults to 'true'")
	private boolean userIsRoot = true;

	@Option(name = "-g", usage = "clean up hdfs dir after export, defaults to 'true'")
	private boolean cleanHdfsDir = true;

	@Option(name = "-c", usage = "compression codec, either 'none', 'gzip' or 'bzip2', defaults to 'gzip'")
	private FileCompressionCodec codec = FileCompressionCodec.gzip;

	@Override
	public int run(String[] args) throws Exception {

		CmdLineParser cmd = new CmdLineParser(this);

		try {
			cmd.parseArgument(args);
		} catch (CmdLineException e) {
			System.err.println(e.getMessage());
			cmd.printUsage(System.err);
			throw e;
		}

		Job job = configure();
		return job.waitForCompletion(true) ? 0 : 1;
	}

	public Job configure(boolean isSecured, String metaStoreUris, String principal,
			String inputDatabase, String inputTable, String inputFilter, int numReducer,
			String[] anonFields, String exportSalt, String keyFile, String ftpUser,
			String ftpPass, String ftpEndpoint, boolean passiveMode, boolean userIsRoot,
			boolean cleanHdfsDir, FileCompressionCodec codec) throws Exception {

		this.isSecured = isSecured;
		this.metaStoreUris = metaStoreUris;
		this.principal = principal;
		this.inputDatabase = inputDatabase;
		this.inputTable = inputTable;
		this.inputFilter = inputFilter;
		this.numReducer = numReducer;
		this.anonFields = anonFields;
		this.exportSalt = exportSalt;
		this.keyFile = keyFile;
		this.ftpUser = ftpUser;
		this.ftpPass = ftpPass;
		this.ftpEndpoint = ftpEndpoint;
		this.passiveMode = passiveMode;
		this.userIsRoot = userIsRoot;
		this.cleanHdfsDir = cleanHdfsDir;
		this.codec = codec;

		return configure();
	}

	private Job configure() throws Exception {

		Configuration conf = getConfiguration();
		conf = configureHiveMetaStore(conf);
		conf = configureKerberos(conf);
		conf = configureAnonFields(conf);

		Job job = Job.getInstance(conf, "FtpExport: " + inputDatabase + "." + inputTable);
{'fraction_non_alphanumeric': 0.11144578313253012, 'fraction_numerical': 0.004381161007667032, 'mean_word_length': 4.6804043545878695, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7663011', 'n_tokens_mistral': 2392, 'n_tokens_neox': 2232, 'n_words': 671}