[go: up one dir, main page]
Skip to content
This repository has been archived by the owner on May 29, 2020. It is now read-only.

修复多线程 parquet sink 的报错问题 #80

Merged
merged 1 commit into from
Jul 21, 2017
Merged

修复多线程 parquet sink 的报错问题 #80

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,22 +124,22 @@
<dependency>
<groupId>org.apache.parquet</groupId>
<artifactId>parquet-common</artifactId>
<version>1.8.1</version>
<version>1.8.2</version>
</dependency>
<dependency>
<groupId>org.apache.parquet</groupId>
<artifactId>parquet-column</artifactId>
<version>1.8.1</version>
<version>1.8.2</version>
</dependency>
<dependency>
<groupId>org.apache.parquet</groupId>
<artifactId>parquet-hadoop</artifactId>
<version>1.8.1</version>
<version>1.8.2</version>
</dependency>
<dependency>
<groupId>org.apache.parquet</groupId>
<artifactId>parquet-avro</artifactId>
<version>1.8.1</version>
<version>1.8.2</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
Expand Down
41 changes: 28 additions & 13 deletions src/main/java/meepo/transform/sink/parquet/ParquetSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@ public class ParquetSink extends AbstractSink {

private ParquetSinkHelper sinkHelper;

private FileSystem hdfs;

public ParquetSink(String name, int index, TaskContext context) {
super(name, index, context);
this.tableName = context.getString("tableName");
Expand Down Expand Up @@ -71,6 +73,13 @@ public void event(DataEvent event) {
@Override
public void onShutdown() {
this.closeParquet();
if (this.hdfs != null) {
try {
this.hdfs.close();
} catch (IOException e) {
e.printStackTrace();
}
}
super.onShutdown();
}

Expand All @@ -81,11 +90,15 @@ public void timeOut() {
public void initHDFS() {
if (this.hdfsConfDir == null)
return;
if (super.indexOfSinks > 0)
try {
this.hdfs = FileSystem.get(createConf(this.hdfsConfDir));
} catch (IOException e) {
e.printStackTrace();
}
if (super.indexOfSinks > 0) {
return;

}
try {
FileSystem hdfs = FileSystem.get(createConf(this.hdfsConfDir));
Path metap = new Path(this.outputDir + "/.metadata");
if (!hdfs.exists(metap)) {
hdfs.mkdirs(metap);
Expand All @@ -97,7 +110,6 @@ public void initHDFS() {
fsd.write(avsc.toString().getBytes(Charset.forName("UTF-8")));
fsd.flush();
fsd.close();
hdfs.close();
}
} catch (Exception e) {
LOG.error("Init HDFS Error :", e);
Expand All @@ -114,7 +126,7 @@ private void initSinkHelper() {
super.RUNNING = true;
this.sinkHelper = (this.hdfsConfDir == null ?
new ParquetSinkHelper(new Path(fileName), this.messageType) :
new ParquetSinkHelper(new Path(fileName), this.messageType, createConf(this.hdfsConfDir)));
new ParquetSinkHelper(this.hdfs.makeQualified(new Path(fileName)), this.messageType, createConf(this.hdfsConfDir)));
} catch (Exception e) {
LOG.error("Init Parquet File Error :", e);
Validate.isTrue(false);
Expand All @@ -125,15 +137,18 @@ private void closeParquet() {
if (this.sinkHelper == null) {
return;
}
try {
this.sinkHelper.close();
this.sinkHelper = null;
this.counter = 0;
this.part++;
super.RUNNING = false;
} catch (IOException e) {
LOG.error("Close Parquet File Error :", e);
while(!this.sinkHelper.isClose()){
try {
this.sinkHelper.close();
} catch (Exception e) {
LOG.error("Close Parquet File Error :", e);
}
Util.sleep(5);
}
this.sinkHelper = null;
this.counter = 0;
this.part++;
super.RUNNING = false;
}

public static Configuration createConf(String classpath) {
Expand Down
14 changes: 10 additions & 4 deletions src/main/java/meepo/transform/sink/parquet/ParquetSinkHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package meepo.transform.sink.parquet;

import lombok.Getter;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.parquet.hadoop.ParquetFileWriter;
Expand All @@ -14,21 +15,26 @@
*/
public class ParquetSinkHelper extends ParquetWriter<Object[]> {

@Getter
private boolean close = false;

@SuppressWarnings("deprecation") public ParquetSinkHelper(Path file, MessageType schema) throws IOException {
@SuppressWarnings("deprecation")
public ParquetSinkHelper(Path file, MessageType schema) throws IOException {
super(file, new ParquetSinkSupport(schema), CompressionCodecName.SNAPPY, ParquetWriter.DEFAULT_BLOCK_SIZE / 4, ParquetWriter.DEFAULT_PAGE_SIZE);
}

@SuppressWarnings("deprecation") public ParquetSinkHelper(Path file, MessageType schema, Configuration conf) throws IOException {
super(file, ParquetFileWriter.Mode.CREATE, new ParquetSinkSupport(schema), CompressionCodecName.SNAPPY, ParquetWriter.DEFAULT_BLOCK_SIZE / 4,
@SuppressWarnings("deprecation")
public ParquetSinkHelper(Path file, MessageType schema, Configuration conf) throws IOException {
super(file, ParquetFileWriter.Mode.CREATE, new ParquetSinkSupport(schema), CompressionCodecName.SNAPPY, ParquetWriter.DEFAULT_BLOCK_SIZE / 8,
ParquetWriter.DEFAULT_PAGE_SIZE, ParquetWriter.DEFAULT_PAGE_SIZE, DEFAULT_IS_DICTIONARY_ENABLED, DEFAULT_IS_VALIDATING_ENABLED, DEFAULT_WRITER_VERSION, conf);
}

@Override public void close() throws IOException {
@Override
public void close() throws IOException {
if (this.close) {
return;
}
super.close();
this.close = true;
}
}