Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add compression config for Parquest files #494

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Add compression config for Parquest files #494

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
96fbdc3
Merge pull request #1 from confluentinc/master
orinciog Apr 6, 2020
403f2fc
Add compression setting for parquet
oranciog-bd Apr 6, 2020
4e847df
Merge branch 'master' of https://github.com/orinciog/kafka-connect-hdfs
oranciog-bd Apr 6, 2020
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
63 changes: 62 additions & 1 deletion src/main/java/io/confluent/connect/hdfs/HdfsSinkConnectorConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,12 +38,15 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import io.confluent.connect.hdfs.avro.AvroFormat;
import io.confluent.connect.hdfs.json.JsonFormat;
import io.confluent.connect.hdfs.storage.HdfsStorage;
import io.confluent.connect.hdfs.parquet.ParquetFormat;
import io.confluent.connect.storage.StorageSinkConnectorConfig;
import io.confluent.connect.storage.common.ComposableConfig;
import io.confluent.connect.storage.common.GenericRecommender;
Expand All @@ -65,6 +68,9 @@
import static io.confluent.connect.storage.common.StorageCommonConfig.TOPICS_DIR_DISPLAY;
import static io.confluent.connect.storage.common.StorageCommonConfig.TOPICS_DIR_DOC;

import org.apache.kafka.common.utils.Utils;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;

public class HdfsSinkConnectorConfig extends StorageSinkConnectorConfig {

// HDFS Group
Expand Down Expand Up @@ -157,7 +163,8 @@ public class HdfsSinkConnectorConfig extends StorageSinkConnectorConfig {
private static final GenericRecommender PARTITIONER_CLASS_RECOMMENDER = new GenericRecommender();
private static final ParentValueRecommender AVRO_COMPRESSION_RECOMMENDER
= new ParentValueRecommender(FORMAT_CLASS_CONFIG, AvroFormat.class, AVRO_SUPPORTED_CODECS);

private static final ParquetCodecRecommender PARQUET_COMPRESSION_RECOMMENDER
= new ParquetCodecRecommender();
static {
STORAGE_CLASS_RECOMMENDER.addValidValues(
Arrays.<Object>asList(HdfsStorage.class)
Expand Down Expand Up @@ -315,6 +322,19 @@ public static ConfigDef newConfigDef() {
hdfsAuthenticationKerberosDependentsRecommender
);
}

final String connectorGroup = "Connector";
final int latestOrderInGroup = configDef.configKeys().values().stream()
.filter(c -> connectorGroup.equalsIgnoreCase(c.group))
.map(c -> c.orderInGroup)
.max(Integer::compare).orElse(0);

StorageSinkConnectorConfig.enableParquetConfig(
configDef,
PARQUET_COMPRESSION_RECOMMENDER,
connectorGroup,
latestOrderInGroup
);
// Put the storage group(s) last ...
ConfigDef storageConfigDef = StorageSinkConnectorConfig.newConfigDef(
FORMAT_CLASS_RECOMMENDER,
Expand Down Expand Up @@ -456,6 +476,12 @@ public Configuration getHadoopConfiguration() {
return hadoopConfig;
}

public CompressionCodecName parquetCompressionCodecName() {
return "none".equalsIgnoreCase(getString(PARQUET_CODEC_CONFIG))
? CompressionCodecName.fromConf(null)
: CompressionCodecName.fromConf(getString(PARQUET_CODEC_CONFIG));
}

public Map<String, ?> plainValues() {
Map<String, Object> map = new HashMap<>();
for (AbstractConfig config : allConfigs) {
Expand Down Expand Up @@ -605,6 +631,41 @@ private void validateReplacements(String config) {
}
}

private static class ParquetCodecRecommender extends ParentValueRecommender
implements ConfigDef.Validator {
public static final Map<String, CompressionCodecName> TYPES_BY_NAME;
public static final List<String> ALLOWED_VALUES;

static {
TYPES_BY_NAME = Arrays.stream(CompressionCodecName.values())
.filter(c -> !CompressionCodecName.UNCOMPRESSED.equals(c))
.collect(Collectors.toMap(c -> c.name().toLowerCase(), Function.identity()));
TYPES_BY_NAME.put("none", CompressionCodecName.UNCOMPRESSED);
ALLOWED_VALUES = new ArrayList<>(TYPES_BY_NAME.keySet());
// Not a hard requirement but this call usually puts 'none' first in the list of allowed
// values
Collections.reverse(ALLOWED_VALUES);
}

public ParquetCodecRecommender() {
super(FORMAT_CLASS_CONFIG, ParquetFormat.class, ALLOWED_VALUES.toArray());
}

@Override
public void ensureValid(String name, Object compressionCodecName) {
String compressionCodecNameString = ((String) compressionCodecName).trim();
if (!TYPES_BY_NAME.containsKey(compressionCodecNameString)) {
throw new ConfigException(name, compressionCodecName,
"Value must be one of: " + ALLOWED_VALUES);
}
}

@Override
public String toString() {
return "[" + Utils.join(ALLOWED_VALUES, ", ") + "]";
}
}

private static class BooleanParentRecommender implements ConfigDef.Recommender {

protected String parentConfigName;
Expand Down
2 changes: 1 addition & 1 deletion src/main/java/io/confluent/connect/hdfs/parquet/ParquetRecordWriterProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ public String getExtension() {
@Override
public RecordWriter getRecordWriter(HdfsSinkConnectorConfig conf, String filename) {
return new RecordWriter() {
final CompressionCodecName compressionCodecName = CompressionCodecName.SNAPPY;
final CompressionCodecName compressionCodecName = conf.parquetCompressionCodecName();
final int blockSize = 256 * 1024 * 1024;
final int pageSize = 64 * 1024;
Path path = new Path(filename);
Expand Down