, FinalState, MetadataAwareWriter, SpeculativeAttemptAwareConstruct {
/**
* {@inheritDoc}.
*
*
* This default implementation simply renames the staging file to the output file. If the output file
* already exists, it will delete it first before doing the renaming.
*
*
* @throws IOException if any file operation fails
*/
@Override
public void commit() throws IOException {
this.closer.close();
setStagingFileGroup();
if (!this.fs.exists(this.stagingFile)) {
throw new IOException(String.format("File %s does not exist", this.stagingFile));
}
FileStatus stagingFileStatus = this.fs.getFileStatus(this.stagingFile);
// Double check permission of staging file
if (!stagingFileStatus.getPermission().equals(this.filePermission)) {
this.fs.setPermission(this.stagingFile, this.filePermission);
}
this.bytesWritten = Optional.of(Long.valueOf(stagingFileStatus.getLen()));
LOG.info(String.format("Moving data from %s to %s", this.stagingFile, this.outputFile));
// For the same reason as deleting the staging file if it already exists, deleting
// the output file if it already exists prevents task retry from being blocked.
if (this.fs.exists(this.outputFile)) {
LOG.warn(String.format("Task output file %s already exists", this.outputFile));
HadoopUtils.deletePath(this.fs, this.outputFile, false);
}
// ⚠️移动stagingFile到outputFile也就是writer的写出文件
HadoopUtils.renamePath(this.fs, this.stagingFile, this.outputFile);
}
}
````
STEP2 写出到最终目录，获取writer写出的文件发布到最终目录
> `public static final String DATA_PUBLISHER_FINAL_DIR = DATA_PUBLISHER_PREFIX + ".final.dir";`
```java
package org.apache.gobblin.publisher;
public class BaseDataPublisher extends SingleTaskDataPublisher {
protected void publishData(WorkUnitState state, int branchId, boolean publishSingleTaskData,
Set writerOutputPathsMoved)
throws IOException {
// Get a ParallelRunner instance for moving files in parallel
ParallelRunner parallelRunner = this.getParallelRunner(this.writerFileSystemByBranches.get(branchId));
// The directory where the workUnitState wrote its output data.
// ⚠️这里就是获取上一步写出的writer文件
Path writerOutputDir = WriterUtils.getWriterOutputDir(state, this.numBranches, branchId);
if (!this.writerFileSystemByBranches.get(branchId).exists(writerOutputDir)) {
LOG.warn(String.format("Branch %d of WorkUnit %s produced no data", branchId, state.getId()));
return;
}
// The directory where the final output directory for this job will be placed.
// It is a combination of DATA_PUBLISHER_FINAL_DIR and WRITER_FILE_PATH.
Path publisherOutputDir = getPublisherOutputDir(state, branchId);
if (publishSingleTaskData) {
// Create final output directory
WriterUtils.mkdirsWithRecursivePermissionWithRetry(this.publisherFileSystemByBranches.get(branchId), publisherOutputDir,
this.permissions.get(branchId), retrierConfig);
addSingleTaskWriterOutputToExistingDir(writerOutputDir, publisherOutputDir, state, branchId, parallelRunner);
} else {
if (writerOutputPathsMoved.contains(writerOutputDir)) {
// This writer output path has already been moved for another task of the same extract
// If publishSingleTaskData=true, writerOutputPathMoved is ignored.
return;
}
if (this.publisherFileSystemByBranches.get(branchId).exists(publisherOutputDir)) {
// The final output directory already exists, check if the job is configured to replace it.
// If publishSingleTaskData=true, final output directory is never replaced.
boolean replaceFinalOutputDir = this.getState().getPropAsBoolean(ForkOperatorUtils
.getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISHER_REPLACE_FINAL_DIR, this.numBranches, branchId));
// If the final output directory is not configured to be replaced, put new data to the existing directory.
if (!replaceFinalOutputDir) {
addWriterOutputToExistingDir(writerOutputDir, publisherOutputDir, state, branchId, parallelRunner);
writerOutputPathsMoved.add(writerOutputDir);
return;
}
// Delete the final output directory if it is configured to be replaced
LOG.info("Deleting publisher output dir " + publisherOutputDir);
this.publisherFileSystemByBranches.get(branchId).delete(publisherOutputDir, true);
} else {
// Create the parent directory of the final output directory if it does not exist
WriterUtils.mkdirsWithRecursivePermissionWithRetry(this.publisherFileSystemByBranches.get(branchId),
publisherOutputDir.getParent(), this.permissions.get(branchId), retrierConfig);
}
// ⚠️移动writerOutputDir到最终输出的publisherOutputDir
movePath(parallelRunner, state, writerOutputDir, publisherOutputDir, branchId);
writerOutputPathsMoved.add(writerOutputDir);
}
}
/**
* close操作将发布的文件路径publisherOutputDirs 附加到ConfigurationKeys.PUBLISHER_DIRS（`data.publisher.output.dirs`）属性的值中，供后续使用，比如 HiveRegistrationPublisher
*/
@Override
public void close()
throws IOException {
try {
for (Path path : this.publisherOutputDirs) {
this.state.appendToSetProp(ConfigurationKeys.PUBLISHER_DIRS, path.toString());
}
} finally {
// 调用google的`com.google.common.io.Closer`这个closer可以注册多个closable(`this.stack.push(closeable)`)在执行close方法时就会调用（`(Closeable)this.stack.pop().close()`）因此可以实现链式的关闭
this.closer.close();
}
}
}
```
STEP4 publish之后的后续publish操作
如果在配置文件中配置了这个publish：
```bash
##配置方案1
data.publisher.type=org.apache.gobblin.publisher.BaseDataPublisherWithHiveRegistration
## 配置方案2
### org.apache.gobblin.runtime.Task.shouldPublishDataInTask()
publish.data.at.job.level=false
job.commit.policy=successful
data.publisher.job.type=org.apache.gobblin.publisher.HiveRegistrationPublisher
data.publisher.task.type=org.apache.gobblin.publisher.BaseDataPublisher
```
```java
package org.apache.gobblin.publisher;
public class BaseDataPublisherWithHiveRegistration extends BaseDataPublisher {
protected final HiveRegistrationPublisher hivePublisher;
public BaseDataPublisherWithHiveRegistration(State state) throws IOException {
super(state);
//向BaseDataPublisher的closer中注册该HiveRegistrationPublisher
this.hivePublisher = this.closer.register(new HiveRegistrationPublisher(state));
}
@Override
public void publish(Collection states) throws IOException {
super.publish(states);
this.hivePublisher.publish(states);
}
}
```
#### Source Extractor Converter 关系
**Source获取Extractor** Source配置`source.class=org.apache.gobblin.source.extractor.extract.kafka.KafkaDeserializerSource`
```java
public class KafkaSimpleSource extends KafkaSource {
@Override
public Extractor getExtractor(WorkUnitState state) throws IOException {
return new KafkaSimpleExtractor(state);
}
}
```
Extractor中配置了`readRecord`方法用于读取数据 Extractor中配置了`getSchema`方法用于获取schema
```java
public class KafkaSimpleExtractor extends KafkaExtractor {
public KafkaSimpleExtractor(WorkUnitState state) {
super(state);
this.kafkaSchemaRegistry = new SimpleKafkaSchemaRegistry(state.getProperties());
}
@Override
public String getSchema() throws IOException {
try {
return this.kafkaSchemaRegistry.getLatestSchemaByTopic(this.topicName);
} catch (SchemaRegistryException e) {
throw new RuntimeException(e);
}
}
}
```
**Converter 转换Schema，Schema从Extractor中获取** Converter配置`converter.classes=org.apache.gobblin.converter.json.JsonStringToJsonIntermediateConverter,org.apache.gobblin.converter.avro.JsonIntermediateToAvroConverter` 这个配置是链式的。按顺序依次转换，[Source Schema and Converters](https://gist.github.com/tilakpatidar/2591c8f4503bcbd0bc0ab212b31ec9b5)
```java
package org.apache.gobblin.runtime;
public class Task implements TaskIFace {
private final Converter converter;
private final InstrumentedExtractorBase extractor;
public Task(TaskContext context, ...) {
//this.taskContext.getExtractor() 的内容: getSource().getExtractor(this.taskState)
this.extractor =
closer.register(new InstrumentedExtractorDecorator<>(this.taskState, this.taskContext.getExtractor()));
}
private void runSynchronousModel() throws Exception {
// 1。 转换schema
Object schema = converter.convertSchema(extractor.getSchema(), this.taskState);
// 2.转换数据
// extractor.readRecordEnvelope():从extractor获取数据
// converter.convertRecord(）:将从extractor获取的数据做转换
RecordEnvelope recordEnvelope;
// Extract, convert, and fork one source record at a time.
while (!shutdownRequested() && (recordEnvelope = extractor.readRecordEnvelope()) != null) {
onRecordExtract();
AcknowledgableWatermark ackableWatermark = new AcknowledgableWatermark(recordEnvelope.getWatermark());
if (watermarkTracker.isPresent()) {
watermarkTracker.get().track(ackableWatermark);
}
for (Object convertedRecord : converter.convertRecord(schema, recordEnvelope, this.taskState)) {
processRecord(convertedRecord, forkOperator, rowChecker, rowResults, branches,
ackableWatermark.incrementAck());
}
ackableWatermark.ack();
}
}
}
```
### workunit的流转
在JobLauncher中调用source组件生成workunit然后生成hadoopjob去执行wu。
```java
package org.apache.gobblin.runtime;
public abstract class AbstractJobLauncher implements JobLauncher {
@Override
public void launchJob(JobListener jobListener) hrows JobException {
//省略...
Source source = this.jobContext.getSource();
//1. 调用source组件生成workunit
if (source instanceof WorkUnitStreamSource) {
workUnitStream = ((WorkUnitStreamSource) source).getWorkunitStream(jobState);
} else {
workUnitStream = new BasicWorkUnitStream.Builder(source.getWorkunits(jobState)).build();
}
//省略...
//2. Start the job and wait for it to finish
// 见下方 MRJobLauncher$runWorkUnits 代码
runWorkUnitStream(workUnitStream);
//省略...
}
}
```
```java
package org.apache.gobblin.runtime.mapreduce;
public class MRJobLauncher extends AbstractJobLauncher {
protected void runWorkUnits(List workUnits) throws Exception {
//省略...
prepareHadoopJob(workUnits);
//省略...
this.job.waitForCompletion(true);
//省略...
}
}
/**
* Prepare the Hadoop MR job, including configuring the job and setting up the input/output paths.
*/
private void prepareHadoopJob(List workUnits) throws IOException {
TimingEvent mrJobSetupTimer = this.eventSubmitter.getTimingEvent(TimingEvent.RunJobTimings.MR_JOB_SETUP);
// Add dependent jars/files
addDependencies(this.job.getConfiguration());
this.job.setJarByClass(MRJobLauncher.class);
this.job.setMapperClass(TaskRunner.class);
// The job is mapper-only
this.job.setNumReduceTasks(0);
this.job.setInputFormatClass(GobblinWorkUnitsInputFormat.class);
this.job.setOutputFormatClass(GobblinOutputFormat.class);
this.job.setMapOutputKeyClass(NullWritable.class);
this.job.setMapOutputValueClass(NullWritable.class);
// Set speculative execution
this.job.setSpeculativeExecution(isSpeculativeExecutionEnabled(this.jobProps));
this.job.getConfiguration().set("mapreduce.job.user.classpath.first", "true");
// Job input path is where input work unit files are stored
// Prepare job input
// wu文件写入hdfs
prepareJobInput(workUnits);
// 设置inputPath
FileInputFormat.addInputPath(this.job, this.jobInputPath);
// Job output path is where serialized task states are stored
FileOutputFormat.setOutputPath(this.job, this.jobOutputPath);
// Serialize source state to a file which will be picked up by the mappers
serializeJobState(this.fs, this.mrJobDir, this.conf, this.jobContext.getJobState(), this.job);
if (this.jobProps.containsKey(ConfigurationKeys.MR_JOB_MAX_MAPPERS_KEY)) {
GobblinWorkUnitsInputFormat.setMaxMappers(this.job,
Integer.parseInt(this.jobProps.getProperty(ConfigurationKeys.MR_JOB_MAX_MAPPERS_KEY)));
}
mrJobSetupTimer.stop();
}
```
### mapreduce相关
**map分块** hadoopjob的`GobblinWorkUnitsInputFormat`中的`getSplits`方法决定了map分块的方式和数量， 在`KafkaSource`中，`getWorkunits`方法是按照partition和`mr.job.max.mappers`参数配置来决定生成的wu的数量， 比如map数量为默认值100，topic的分区数量为2，那么就会生成100个multiworkunit，其中有两个是有job的，对应topic的两个分区，其余的job均为空，如果分区数量大于map数量，那么其中便会有一些multiworkunit去获取多个分区的数据。
```java
package org.apache.gobblin.source.extractor.extract.kafka;
public abstract class KafkaSource extends EventBasedSource {
public List getWorkunits(SourceState state) {
//省略...
Map> workUnits = Maps.newConcurrentMap();
//根据topic创建WorkUnitCreator
for (KafkaTopic topic : topics) {
threadPool.submit(
new WorkUnitCreator(topic, state, Optional.fromNullable(topicSpecificStateMap.get(topic.getName())),
workUnits));
}
//省略...
//获取map数量
int numOfMultiWorkunits =
state.getPropAsInt(ConfigurationKeys.MR_JOB_MAX_MAPPERS_KEY, ConfigurationKeys.DEFAULT_MR_JOB_MAX_MAPPERS);
//KafkaWorkUnitPacker是用来合并空的任务以及将workunit分配到numOfMultiWorkunits指定的数量的map上，见KafkaSingleLevelWorkUnitPacker
List workUnitList = KafkaWorkUnitPacker.getInstance(this, state).pack(workUnits, numOfMultiWorkunits);
//省略...
}
private class WorkUnitCreator implements Runnable {
@Override
public void run() {
//省略...
//生成workunit
this.allTopicWorkUnits.put(this.topic.getName(),
KafkaSource.this.getWorkUnitsForTopic(this.topic, this.state, this.topicSpecificState));
//省略...
}
}
private List getWorkUnitsForTopic(KafkaTopic topic, SourceState state, Optional topicSpecificState) {
Timer.Context context = this.metricContext.timer("isTopicQualifiedTimer").time();
boolean topicQualified = isTopicQualified(topic);
context.close();
List workUnits = Lists.newArrayList();
for (KafkaPartition partition : topic.getPartitions()) {
WorkUnit workUnit = getWorkUnitForTopicPartition(partition, state, topicSpecificState);
this.partitionsToBeProcessed.add(partition);
if (workUnit != null) {
// For disqualified topics, for each of its workunits set the high watermark to be the same
// as the low watermark, so that it will be skipped.
if (!topicQualified) {
skipWorkUnit(workUnit);
}
workUnits.add(workUnit);
}
}
return workUnits;
}
}
```
# sqoop 简介及使用
## 1. 前言
Sqoop是用于在Hadoop和关系型数据库之间流转数据的工具。可以使用Sqoop从关系型数据库（RDBMS）比如Mysql或者Oracle导入数据到Hadoop分布式文件系统上，在Hadoop MapReduce上对数据做转换，然后导回到RDBMS上。 借助数据库的Schema描述信息，Sqoop自动实现了上述的很多过程，Sqoop使用MapReduce去导入和导出数据，这样可以提供并行操作以及提高容错能力。
## 2. Sqoop的Shell（命令行客户端）操作
```
usage: sqoop COMMAND [ARGS]
Available commands:
codegen 生成Java代码
create-hive-table 根据表结构生成hive表
eval 执行SQL语句并显示结果
export 导出HDFS目录到数据库表
help 帮助
import 从数据库导入数据到HDFS
import-all-tables 导入数据库所有表到HDFS
list-databases 列举所有的database
list-tables 列举数据库中的所有表
version 查看版本信息
```
### 2.1 sqoop import
#### 2.1.1 命令行客户端支持的命令参数
```
usage: sqoop import [GENERIC-ARGS] [TOOL-ARGS]
Common arguments:
--connect Specify JDBC
connect
string
--connection-manager Specify
connection
manager
class name
--connection-param-file Specify
connection
parameters
file
--driver Manually
specify JDBC
driver class
to use
--hadoop-home Override
$HADOOP_MAPR
ED_HOME_ARG
--hadoop-mapred-home Override
$HADOOP_MAPR
ED_HOME_ARG
--help Print usage
instructions
--metadata-transaction-isolation-level Defines the
transaction
isolation
level for
metadata
queries. For
more details
check
java.sql.Con
nection
javadoc or
the JDBC
specificaito
n
--oracle-escaping-disabled Disable the
escaping
mechanism of
the
Oracle/OraOo
p connection
managers
-P Read
password
from console
--password Set
authenticati
on password
--password-alias Credential
provider
password
alias
--password-file Set
authenticati
on password
file path
--relaxed-isolation Use
read-uncommi
tted
isolation
for imports
--skip-dist-cache Skip copying
jars to
distributed
cache
--temporary-rootdir Defines the
temporary
root
directory
for the
import
--throw-on-error Rethrow a
RuntimeExcep
tion on
error
occurred
during the
job
--username Set
authenticati
on username
--verbose Print more
information
while
working
Import control arguments:
--append Imports data
in append
mode
--as-avrodatafile Imports data
to Avro data
files
--as-parquetfile Imports data
to Parquet
files
--as-sequencefile Imports data
to
SequenceFile
s
--as-textfile Imports data
as plain
text
(default)
--autoreset-to-one-mapper Reset the
number of
mappers to
one mapper
if no split
key
available
--boundary-query Set boundary
query for
retrieving
max and min
value of the
primary key
--columns Columns to
import from
table
--compression-codec Compression
codec to use
for import
--delete-target-dir Imports data
in delete
mode
--direct Use direct
import fast
path
--direct-split-size Split the
input stream
every 'n'
bytes when
importing in
direct mode
-e,--query Import
results of
SQL
'statement'
--fetch-size Set number
'n' of rows
to fetch
from the
database
when more
rows are
needed
--inline-lob-limit Set the
maximum size
for an
inline LOB
-m,--num-mappers Use 'n' map
tasks to
import in
parallel
--mapreduce-job-name Set name for
generated
mapreduce
job
--merge-key Key column
to use to
join results
--split-by Column of
the table
used to
split work
units
--split-limit Upper Limit
of rows per
split for
split
columns of
Date/Time/Ti
mestamp and
integer
types. For
date or
timestamp
fields it is
calculated
in seconds.
split-limit
should be
greater than
0
--table Table to
read
--target-dir HDFS plain
table
destination
--validate Validate the
copy using
the
configured
validator
--validation-failurehandler Fully
qualified
class name
for
ValidationFa
ilureHandler
--validation-threshold Fully
qualified
class name
for
ValidationTh
reshold
--validator Fully
qualified
class name
for the
Validator
--warehouse-dir HDFS parent
for table
destination
--where WHERE clause
to use
during
import
-z,--compress Enable
compression
Incremental import arguments:
--check-column Source column to check for incremental
change
--incremental Define an incremental import of type
'append' or 'lastmodified'
--last-value Last imported value in the incremental
check column
Output line formatting arguments:
--enclosed-by Sets a required field enclosing
character
--escaped-by Sets the escape character
--fields-terminated-by Sets the field separator character
--lines-terminated-by Sets the end-of-line character
--mysql-delimiters Uses MySQL's default delimiter set:
fields: , lines: \n escaped-by: \
optionally-enclosed-by: '
--optionally-enclosed-by Sets a field enclosing character
Input parsing arguments:
--input-enclosed-by Sets a required field encloser
--input-escaped-by Sets the input escape
character
--input-fields-terminated-by Sets the input field separator
--input-lines-terminated-by Sets the input end-of-line
char
--input-optionally-enclosed-by Sets a field enclosing
character
Hive arguments:
--create-hive-table Fail if the target hive
table exists
--external-table-dir Sets where the external
table is in HDFS
--hive-database Sets the database name to
use when importing to hive
--hive-delims-replacement Replace Hive record \0x01
and row delimiters (\n\r)
from imported string fields
with user-defined string
--hive-drop-import-delims Drop Hive record \0x01 and
row delimiters (\n\r) from
imported string fields
--hive-home Override $HIVE_HOME
--hive-import Import tables into Hive
(Uses Hive's default
delimiters if none are
set.)
--hive-overwrite Overwrite existing data in
the Hive table
--hive-partition-key Sets the partition key to
use when importing to hive
--hive-partition-value Sets the partition value to
use when importing to hive
--hive-table Sets the table name to use
when importing to hive
--map-column-hive Override mapping for
specific column to hive
types.
HBase arguments:
--column-family Sets the target column family for the
import
--hbase-bulkload Enables HBase bulk loading
--hbase-create-table If specified, create missing HBase tables
--hbase-row-key Specifies which input column to use as the
row key
--hbase-table Import to in HBase
HCatalog arguments:
--hcatalog-database HCatalog database name
--hcatalog-home Override $HCAT_HOME
--hcatalog-partition-keys Sets the partition
keys to use when
importing to hive
--hcatalog-partition-values Sets the partition
values to use when
importing to hive
--hcatalog-table HCatalog table name
--hive-home Override $HIVE_HOME
--hive-partition-key Sets the partition key
to use when importing
to hive
--hive-partition-value Sets the partition
value to use when
importing to hive
--map-column-hive Override mapping for
specific column to
hive types.
HCatalog import specific options:
--create-hcatalog-table Create HCatalog before import
--drop-and-create-hcatalog-table Drop and Create HCatalog before
import
--hcatalog-storage-stanza HCatalog storage stanza for table
creation
Accumulo arguments:
--accumulo-batch-size Batch size in bytes
--accumulo-column-family Sets the target column family for
the import
--accumulo-create-table If specified, create missing
Accumulo tables
--accumulo-instance Accumulo instance name.
--accumulo-max-latency Max write latency in milliseconds
--accumulo-password Accumulo password.
--accumulo-row-key Specifies which input column to
use as the row key
--accumulo-table Import to in Accumulo
--accumulo-user Accumulo user name.
--accumulo-visibility Visibility token to be applied to
all rows imported
--accumulo-zookeepers Comma-separated list of
zookeepers (host:port)
Code generation arguments:
--bindir Output directory for
compiled objects
--class-name Sets the generated class
name. This overrides
--package-name. When
combined with --jar-file,
sets the input class.
--escape-mapping-column-names Disable special characters
escaping in column names
--input-null-non-string Input null non-string
representation
--input-null-string Input null string
representation
--jar-file Disable code generation; use
specified jar
--map-column-java Override mapping for
specific columns to java
types
--null-non-string Null non-string
representation
--null-string Null string representation
--outdir Output directory for
generated code
--package-name Put auto-generated classes
in this package
Generic Hadoop command-line arguments:
(must preceed any tool-specific arguments)
Generic options supported are
-conf specify an application configuration file
-D use value for given property
-fs specify a namenode
-jt specify a ResourceManager
-files specify comma separated files to be copied to the map reduce cluster
-libjars specify comma separated jar files to include in the classpath.
-archives specify comma separated archives to be unarchived on the compute machines.
The general command line syntax is
bin/hadoop command [genericOptions] [commandOptions]
At minimum, you must specify --connect and --table
Arguments to mysqldump and other subprograms may be supplied
after a '--' on the command line.
```
#### 2.1.2 公共参数
`--connect ` 制定JDBC连接串
`--connection-manager ` 指定连接管理类名
`--driver ` 指定JDBC驱动类
`--hadoop-mapred-home ` 覆盖$HADOOP\_MAPRED\_HOME
`--help` 帮助
`--password-file` 指定密码文件
`-P` 从命令行读取密码
`--password ` 指定密码
`--username ` 指定用户名
`--verbose` 显示更丰富的日志
`--connection-param-file ` 连接参数的可选配置文件
#### 2.1.3 连接一个数据库
使用`--conenct`来连接数据库\
`$ sqoop import --connect jdbc:mysql://database.example.com/employees`
这个连接串会连接database.example.com服务器上的employees数据库。必须要注意如果你准备在分布式Hadoop集群使用Sqoop导入数据，你不能使用`localhost`作为Url来连接数据源，因为这个连接串会被应用于你的MapReduce集群的任务节点上。\
指定数据库连接后，需要提供数据库访问的用户名和密码，密码有多种设定方式，显式的指定密码是不安全的
```bash
$ sqoop import --connect jdbc:mysql://database.example.com/employees \
--username venkatesh --password-file ${user.home}/.password
```
Sqoop内置了MySql的驱动，对于其他的，需要自己安装驱动，将驱动文件放到`$SQOOP_HOME/lib`目录下，然后使用`--driver`参数指定驱动的全类名
```bash
$ sqoop import --driver com.microsoft.jdbc.sqlserver.SQLServerDriver \
--connect ...
```
当使用JDBC连接一个数据库时，也可以使用`--connection-param-file`参数来通过一个配置文件指定额外的参数，这个文件会被作为标准的Java配置来解析并传递给驱动来创建连接。
#### 2.1.4 Import 配置参数
`--append` 追加数据到HDFS中已有的数据文件
`--as-avrodatafile` 导入数据为Avro格式
`--as-sequencefile` 导入数据为SqeuqnceFile
`--as-textfile` 导入数据格式为textfile(默认)
`--as-parquetfile` 导入数据为Parquet格式
`--boundary-query ` 覆盖创建分割时的边界查询SQL
`--columns ` 指定导入列
`--delete-target-dir` 如果存在，删除导入的target目录
`--direct` 如果数据源支持，使用direct模式导入
`-m,--num-mappers ` 指定并行导入数据的map数量(正整数)，默认为4个
`-e,--query ` 导入查询语句得到的数据
`--split-by ` 指定分割split的字段
`--table < table-name >` 制定数据库中的表名
`--target-dir < dir >` 目标HDFS路径
`--warehouse-dir ` 表目标位置父路径
`--where ` 导入的过滤条件，即sql语句的where条件
`--null-string < null-string >` 源表中为null的记录导入为string类型时显示为null-string，默认显示为”null”
`--null-non-string ` 源表中为null的记录导入为非string类型时显示为null-string，默认显示为”null”
### 2.2 sqoop export
#### 2.2.1 命令行客户端支持的命令参数
```
usage: sqoop export [GENERIC-ARGS] [TOOL-ARGS]
Common arguments:
--connect Specify JDBC
connect
string
--connection-manager Specify
connection
manager
class name
--connection-param-file Specify
connection
parameters
file
--driver Manually
specify JDBC
driver class
to use
--hadoop-home Override
$HADOOP_MAPR
ED_HOME_ARG
--hadoop-mapred-home Override
$HADOOP_MAPR
ED_HOME_ARG
--help Print usage
instructions
--metadata-transaction-isolation-level Defines the
transaction
isolation
level for
metadata
queries. For
more details
check
java.sql.Con
nection
javadoc or
the JDBC
specificaito
n
--oracle-escaping-disabled Disable the
escaping
mechanism of
the
Oracle/OraOo
p connection
managers
-P Read
password
from console
--password Set
authenticati
on password
--password-alias Credential
provider
password
alias
--password-file Set
authenticati
on password
file path
--relaxed-isolation Use
read-uncommi
tted
isolation
for imports
--skip-dist-cache Skip copying
jars to
distributed
cache
--temporary-rootdir Defines the
temporary
root
directory
for the
import
--throw-on-error Rethrow a
RuntimeExcep
tion on
error
occurred
during the
job
--username Set
authenticati
on username
--verbose Print more
information
while
working
Export control arguments:
--batch Indicates
underlying
statements
to be
executed in
batch mode
--call Populate the
table using
this stored
procedure
(one call
per row)
--clear-staging-table Indicates
that any
data in
staging
table can be
deleted
--columns Columns to
export to
table
--direct Use direct
export fast
path
--export-dir HDFS source
path for the
export
-m,--num-mappers Use 'n' map
tasks to
export in
parallel
--mapreduce-job-name Set name for
generated
mapreduce
job
--staging-table Intermediate
staging
table
--table Table to
populate
--update-key Update
records by
specified
key column
--update-mode Specifies
how updates
are
performed
when new
rows are
found with
non-matching
keys in
database
--validate Validate the
copy using
the
configured
validator
--validation-failurehandler Fully
qualified
class name
for
ValidationFa
ilureHandler
--validation-threshold Fully
qualified
class name
for
ValidationTh
reshold
--validator Fully
qualified
class name
for the
Validator
Input parsing arguments:
--input-enclosed-by Sets a required field encloser
--input-escaped-by Sets the input escape
character
--input-fields-terminated-by Sets the input field separator
--input-lines-terminated-by Sets the input end-of-line
char
--input-optionally-enclosed-by Sets a field enclosing
character
Output line formatting arguments:
--enclosed-by Sets a required field enclosing
character
--escaped-by Sets the escape character
--fields-terminated-by Sets the field separator character
--lines-terminated-by Sets the end-of-line character
--mysql-delimiters Uses MySQL's default delimiter set:
fields: , lines: \n escaped-by: \
optionally-enclosed-by: '
--optionally-enclosed-by Sets a field enclosing character
Code generation arguments:
--bindir Output directory for
compiled objects
--class-name Sets the generated class
name. This overrides
--package-name. When
combined with --jar-file,
sets the input class.
--escape-mapping-column-names Disable special characters
escaping in column names
--input-null-non-string Input null non-string
representation
--input-null-string Input null string
representation
--jar-file Disable code generation; use
specified jar
--map-column-java Override mapping for
specific columns to java
types
--null-non-string Null non-string
representation
--null-string Null string representation
--outdir Output directory for
generated code
--package-name Put auto-generated classes
in this package
HCatalog arguments:
--hcatalog-database HCatalog database name
--hcatalog-home Override $HCAT_HOME
--hcatalog-partition-keys Sets the partition
keys to use when
importing to hive
--hcatalog-partition-values Sets the partition
values to use when
importing to hive
--hcatalog-table HCatalog table name
--hive-home Override $HIVE_HOME
--hive-partition-key Sets the partition key
to use when importing
to hive
--hive-partition-value Sets the partition
value to use when
importing to hive
--map-column-hive