| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You 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.apache.gobblin.iceberg.writer; |
| |
| import java.io.IOException; |
| import java.time.Instant; |
| import java.time.LocalDate; |
| import java.time.ZoneId; |
| import java.time.ZonedDateTime; |
| import java.time.format.DateTimeFormatter; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.Comparator; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.SortedSet; |
| import java.util.TreeSet; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.locks.Lock; |
| import java.util.concurrent.locks.ReadWriteLock; |
| import java.util.concurrent.locks.ReentrantReadWriteLock; |
| import java.util.function.Consumer; |
| import java.util.stream.Collectors; |
| import java.util.stream.Stream; |
| |
| import org.apache.avro.generic.GenericRecord; |
| import org.apache.avro.specific.SpecificData; |
| |
| import org.apache.gobblin.hive.writer.MetadataWriterKeys; |
| import org.apache.gobblin.source.extractor.extract.LongWatermark; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.fs.permission.FsPermission; |
| import org.apache.iceberg.AppendFiles; |
| import org.apache.iceberg.DataFile; |
| import org.apache.iceberg.DeleteFiles; |
| import org.apache.iceberg.ExpireSnapshots; |
| import org.apache.iceberg.FindFiles; |
| import org.apache.iceberg.PartitionSpec; |
| import org.apache.iceberg.Schema; |
| import org.apache.iceberg.Snapshot; |
| import org.apache.iceberg.StructLike; |
| import org.apache.iceberg.Table; |
| import org.apache.iceberg.TableProperties; |
| import org.apache.iceberg.Transaction; |
| import org.apache.iceberg.UpdateProperties; |
| import org.apache.iceberg.avro.AvroSchemaUtil; |
| import org.apache.iceberg.catalog.TableIdentifier; |
| import org.apache.iceberg.exceptions.AlreadyExistsException; |
| import org.apache.iceberg.exceptions.NoSuchTableException; |
| import org.apache.iceberg.expressions.Expression; |
| import org.apache.iceberg.expressions.Expressions; |
| import org.apache.iceberg.hive.HiveCatalog; |
| import org.apache.iceberg.hive.HiveCatalogs; |
| import org.apache.iceberg.types.Types; |
| import org.joda.time.DateTime; |
| import org.joda.time.format.PeriodFormatter; |
| import org.joda.time.format.PeriodFormatterBuilder; |
| |
| import com.codahale.metrics.Timer; |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.base.Joiner; |
| import com.google.common.base.Optional; |
| import com.google.common.base.Preconditions; |
| import com.google.common.base.Splitter; |
| import com.google.common.cache.Cache; |
| import com.google.common.cache.CacheBuilder; |
| import com.google.common.collect.Lists; |
| import com.google.common.collect.Maps; |
| import com.google.common.collect.Range; |
| import com.google.common.collect.Sets; |
| import com.google.common.io.Closer; |
| |
| import lombok.Getter; |
| import lombok.Setter; |
| import lombok.extern.slf4j.Slf4j; |
| |
| import org.apache.gobblin.completeness.verifier.KafkaAuditCountVerifier; |
| import org.apache.gobblin.configuration.ConfigurationKeys; |
| import org.apache.gobblin.configuration.State; |
| import org.apache.gobblin.data.management.copy.hive.WhitelistBlacklist; |
| import org.apache.gobblin.hive.AutoCloseableHiveLock; |
| import org.apache.gobblin.hive.HiveLock; |
| import org.apache.gobblin.hive.HivePartition; |
| import org.apache.gobblin.hive.metastore.HiveMetaStoreUtils; |
| import org.apache.gobblin.hive.spec.HiveSpec; |
| import org.apache.gobblin.hive.writer.MetadataWriter; |
| import org.apache.gobblin.iceberg.Utils.IcebergUtils; |
| import org.apache.gobblin.metadata.GobblinMetadataChangeEvent; |
| import org.apache.gobblin.metadata.OperationType; |
| import org.apache.gobblin.metrics.GobblinMetricsRegistry; |
| import org.apache.gobblin.metrics.MetricContext; |
| import org.apache.gobblin.metrics.Tag; |
| import org.apache.gobblin.metrics.event.EventSubmitter; |
| import org.apache.gobblin.metrics.event.GobblinEventBuilder; |
| import org.apache.gobblin.metrics.kafka.KafkaSchemaRegistry; |
| import org.apache.gobblin.metrics.kafka.SchemaRegistryException; |
| import org.apache.gobblin.stream.RecordEnvelope; |
| import org.apache.gobblin.time.TimeIterator; |
| import org.apache.gobblin.util.AvroUtils; |
| import org.apache.gobblin.util.ClustersNames; |
| import org.apache.gobblin.util.HadoopUtils; |
| import org.apache.gobblin.util.ParallelRunner; |
| import org.apache.gobblin.util.WriterUtils; |
| import static org.apache.gobblin.iceberg.writer.IcebergMetadataWriterConfigKeys.*; |
| |
| /** |
| * This writer is used to calculate iceberg metadata from GMCE and register to iceberg |
| * iceberg metadata here includes: |
| * 1. Data files that contained by the table and the metrics of the data files |
| * 2. Properties of the table (origin table properties, data offset range, high watermark of the GMCE and schema created time) |
| * 3. Latest schema of the table |
| */ |
| @Slf4j |
| public class IcebergMetadataWriter implements MetadataWriter { |
| |
| // Critical when there's dataset-level ACL enforced for both data and Iceberg metadata |
| public static final String USE_DATA_PATH_AS_TABLE_LOCATION = "use.data.path.as.table.location"; |
| public static final String TABLE_LOCATION_SUFFIX = "/_iceberg_metadata/%s"; |
| public static final String GMCE_HIGH_WATERMARK_KEY = "gmce.high.watermark.%s"; |
| public static final String GMCE_LOW_WATERMARK_KEY = "gmce.low.watermark.%s"; |
| private final static String EXPIRE_SNAPSHOTS_LOOKBACK_TIME = "gobblin.iceberg.dataset.expire.snapshots.lookBackTime"; |
| private final static String DEFAULT_EXPIRE_SNAPSHOTS_LOOKBACK_TIME = "3d"; |
| private static final String ICEBERG_REGISTRATION_BLACKLIST = "iceberg.registration.blacklist"; |
| private static final String ICEBERG_REGISTRATION_WHITELIST = "iceberg.registration.whitelist"; |
| private static final String ICEBERG_REGISTRATION_AUDIT_COUNT_BLACKLIST = "iceberg.registration.audit.count.blacklist"; |
| private static final String ICEBERG_REGISTRATION_AUDIT_COUNT_WHITELIST = "iceberg.registration.audit.count.whitelist"; |
| private static final String ICEBERG_METADATA_FILE_PERMISSION = "iceberg.metadata.file.permission"; |
| private static final String CREATE_TABLE_TIME = "iceberg.create.table.time"; |
| private static final String SCHEMA_CREATION_TIME_KEY = "schema.creation.time"; |
| private static final String ADDED_FILES_CACHE_EXPIRING_TIME = "added.files.cache.expiring.time"; |
| private static final int DEFAULT_ADDED_FILES_CACHE_EXPIRING_TIME = 1; |
| private static final String OFFSET_RANGE_KEY_PREFIX = "offset.range."; |
| private static final String OFFSET_RANGE_KEY_FORMAT = OFFSET_RANGE_KEY_PREFIX + "%s"; |
| |
| private static final String DEFAULT_CREATION_TIME = "0"; |
| private static final String SNAPSHOT_EXPIRE_THREADS = "snapshot.expire.threads"; |
| private static final long DEFAULT_WATERMARK = -1L; |
| |
| /* one of the fields in DataFile entry to describe the location URI of a data file with FS Scheme */ |
| private static final String ICEBERG_FILE_PATH_COLUMN = DataFile.FILE_PATH.name(); |
| |
| private final boolean completenessEnabled; |
| private final WhitelistBlacklist completenessWhitelistBlacklist; |
| private final String timeZone; |
| private final DateTimeFormatter HOURLY_DATEPARTITION_FORMAT; |
| private final String newPartitionColumn; |
| private final String newPartitionColumnType; |
| private final boolean newPartitionEnabled; |
| private final WhitelistBlacklist newPartitionTableWhitelistBlacklist; |
| private Optional<KafkaAuditCountVerifier> auditCountVerifier; |
| private final String auditCheckGranularity; |
| |
| protected final MetricContext metricContext; |
| protected EventSubmitter eventSubmitter; |
| private final WhitelistBlacklist whitelistBlacklist; |
| private final WhitelistBlacklist auditWhitelistBlacklist; |
| private final Closer closer = Closer.create(); |
| |
| // Mapping between table-id and currently processed watermark |
| private final Map<TableIdentifier, Long> tableCurrentWatermarkMap; |
| // Used to store the relationship between table and the gmce topicPartition |
| private final Map<TableIdentifier, String> tableTopicPartitionMap; |
| @Getter |
| private final KafkaSchemaRegistry schemaRegistry; |
| private final Map<TableIdentifier, TableMetadata> tableMetadataMap; |
| @Setter |
| protected HiveCatalog catalog; |
| protected final Configuration conf; |
| protected final ReadWriteLock readWriteLock; |
| private final HiveLock locks; |
| private final boolean useDataLocationAsTableLocation; |
| private final ParallelRunner parallelRunner; |
| private FsPermission permission; |
| protected State state; |
| |
| public IcebergMetadataWriter(State state) throws IOException { |
| this.state = state; |
| this.schemaRegistry = KafkaSchemaRegistry.get(state.getProperties()); |
| conf = HadoopUtils.getConfFromState(state); |
| initializeCatalog(); |
| tableTopicPartitionMap = new HashMap<>(); |
| tableMetadataMap = new HashMap<>(); |
| tableCurrentWatermarkMap = new HashMap<>(); |
| List<Tag<?>> tags = Lists.newArrayList(); |
| String clusterIdentifier = ClustersNames.getInstance().getClusterName(); |
| tags.add(new Tag<>(MetadataWriterKeys.CLUSTER_IDENTIFIER_KEY_NAME, clusterIdentifier)); |
| metricContext = closer.register( |
| GobblinMetricsRegistry.getInstance().getMetricContext(state, IcebergMetadataWriter.class, tags)); |
| this.eventSubmitter = |
| new EventSubmitter.Builder(this.metricContext, MetadataWriterKeys.METRICS_NAMESPACE_ICEBERG_WRITER).build(); |
| this.whitelistBlacklist = new WhitelistBlacklist(state.getProp(ICEBERG_REGISTRATION_WHITELIST, ""), |
| state.getProp(ICEBERG_REGISTRATION_BLACKLIST, "")); |
| this.auditWhitelistBlacklist = new WhitelistBlacklist(state.getProp(ICEBERG_REGISTRATION_AUDIT_COUNT_WHITELIST, ""), |
| state.getProp(ICEBERG_REGISTRATION_AUDIT_COUNT_BLACKLIST, "")); |
| |
| // Use rw-lock to make it thread-safe when flush and write(which is essentially aggregate & reading metadata), |
| // are called in separate threads. |
| readWriteLock = new ReentrantReadWriteLock(); |
| this.locks = new HiveLock(state.getProperties()); |
| parallelRunner = closer.register(new ParallelRunner(state.getPropAsInt(SNAPSHOT_EXPIRE_THREADS, 20), |
| FileSystem.get(HadoopUtils.getConfFromState(state)))); |
| useDataLocationAsTableLocation = state.getPropAsBoolean(USE_DATA_PATH_AS_TABLE_LOCATION, false); |
| if (useDataLocationAsTableLocation) { |
| permission = |
| HadoopUtils.deserializeFsPermission(state, ICEBERG_METADATA_FILE_PERMISSION, |
| FsPermission.getDefault()); |
| } |
| this.completenessEnabled = state.getPropAsBoolean(ICEBERG_COMPLETENESS_ENABLED, DEFAULT_ICEBERG_COMPLETENESS); |
| this.completenessWhitelistBlacklist = new WhitelistBlacklist(state.getProp(ICEBERG_COMPLETENESS_WHITELIST, ""), |
| state.getProp(ICEBERG_COMPLETENESS_BLACKLIST, "")); |
| this.timeZone = state.getProp(TIME_ZONE_KEY, DEFAULT_TIME_ZONE); |
| this.HOURLY_DATEPARTITION_FORMAT = DateTimeFormatter.ofPattern(DATEPARTITION_FORMAT) |
| .withZone(ZoneId.of(this.timeZone)); |
| this.auditCountVerifier = Optional.fromNullable(this.completenessEnabled ? new KafkaAuditCountVerifier(state) : null); |
| this.newPartitionColumn = state.getProp(NEW_PARTITION_KEY, DEFAULT_NEW_PARTITION); |
| this.newPartitionColumnType = state.getProp(NEW_PARTITION_TYPE_KEY, DEFAULT_PARTITION_COLUMN_TYPE); |
| this.newPartitionEnabled = state.getPropAsBoolean(ICEBERG_NEW_PARTITION_ENABLED, DEFAULT_ICEBERG_NEW_PARTITION_ENABLED); |
| this.newPartitionTableWhitelistBlacklist = new WhitelistBlacklist(state.getProp(ICEBERG_NEW_PARTITION_WHITELIST, ""), |
| state.getProp(ICEBERG_NEW_PARTITION_BLACKLIST, "")); |
| this.auditCheckGranularity = state.getProp(AUDIT_CHECK_GRANULARITY, DEFAULT_AUDIT_CHECK_GRANULARITY); |
| } |
| |
| @VisibleForTesting |
| protected void setAuditCountVerifier(KafkaAuditCountVerifier verifier) { |
| this.auditCountVerifier = Optional.of(verifier); |
| } |
| |
| protected void initializeCatalog() { |
| catalog = HiveCatalogs.loadCatalog(conf); |
| } |
| |
| private org.apache.iceberg.Table getIcebergTable(TableIdentifier tid) throws NoSuchTableException { |
| TableMetadata tableMetadata = tableMetadataMap.computeIfAbsent(tid, t -> new TableMetadata()); |
| if (!tableMetadata.table.isPresent()) { |
| tableMetadata.table = Optional.of(catalog.loadTable(tid)); |
| } |
| return tableMetadata.table.get(); |
| } |
| |
| /** |
| * The method is used to get current watermark of the gmce topic partition for a table, and persist the value |
| * in the {@link #tableMetadataMap} as a side effect. |
| * |
| * Make the watermark config name contains topicPartition in case we change the gmce topic name for some reason |
| */ |
| private Long getAndPersistCurrentWatermark(TableIdentifier tid, String topicPartition) { |
| if (tableCurrentWatermarkMap.containsKey(tid)) { |
| return tableCurrentWatermarkMap.get(tid); |
| } |
| org.apache.iceberg.Table icebergTable; |
| Long currentWatermark = DEFAULT_WATERMARK; |
| try { |
| icebergTable = getIcebergTable(tid); |
| } catch (NoSuchTableException e) { |
| return currentWatermark; |
| } |
| currentWatermark = |
| icebergTable.properties().containsKey(String.format(GMCE_HIGH_WATERMARK_KEY, topicPartition)) ? Long.parseLong( |
| icebergTable.properties().get(String.format(GMCE_HIGH_WATERMARK_KEY, topicPartition))) : DEFAULT_WATERMARK; |
| return currentWatermark; |
| } |
| |
| /** |
| * The write method will be responsible for processing gmce and aggregating the metadata. |
| * The logic of this function will be: |
| * 1. Check whether a table exists, if not then create a iceberg table |
| * - If completeness is enabled, Add new parititon column to |
| * table {#NEW_PARTITION_KEY} |
| * 2. Compute schema from the gmce and update the cache for candidate schemas |
| * 3. Do the required operation of the gmce, i.e. addFile, rewriteFile, dropFile or change_property. |
| * |
| * Note: this method only aggregates the metadata in cache without committing, |
| * while the actual commit will be done in flush method (except rewrite and drop methods where preserving older file |
| * information increases the memory footprints, therefore we would like to flush them eagerly). |
| */ |
| public void write(GobblinMetadataChangeEvent gmce, Map<String, Collection<HiveSpec>> newSpecsMap, |
| Map<String, Collection<HiveSpec>> oldSpecsMap, HiveSpec tableSpec) throws IOException { |
| TableIdentifier tid = TableIdentifier.of(tableSpec.getTable().getDbName(), tableSpec.getTable().getTableName()); |
| TableMetadata tableMetadata = tableMetadataMap.computeIfAbsent(tid, t -> new TableMetadata()); |
| Table table; |
| try { |
| table = getIcebergTable(tid); |
| } catch (NoSuchTableException e) { |
| try { |
| if (gmce.getOperationType() == OperationType.drop_files || |
| gmce.getOperationType() == OperationType.change_property) { |
| log.warn("Table {} does not exist, skip processing this {} event", tid.toString(), gmce.getOperationType()); |
| return; |
| } |
| table = createTable(gmce, tableSpec); |
| tableMetadata.table = Optional.of(table); |
| } catch (Exception e1) { |
| log.error("skip processing {} for table {}.{} due to error when creating table", gmce.toString(), |
| tableSpec.getTable().getDbName(), tableSpec.getTable().getTableName()); |
| log.debug(e1.toString()); |
| return; |
| } |
| } |
| if(tableMetadata.completenessEnabled) { |
| tableMetadata.completionWatermark = Long.parseLong(table.properties().getOrDefault(COMPLETION_WATERMARK_KEY, |
| String.valueOf(DEFAULT_COMPLETION_WATERMARK))); |
| } |
| |
| computeCandidateSchema(gmce, tid, tableSpec); |
| tableMetadata.ensureTxnInit(); |
| tableMetadata.lowestGMCEEmittedTime = Long.min(tableMetadata.lowestGMCEEmittedTime, gmce.getGMCEmittedTime()); |
| switch (gmce.getOperationType()) { |
| case add_files: { |
| updateTableProperty(tableSpec, tid); |
| addFiles(gmce, newSpecsMap, table, tableMetadata); |
| if (gmce.getAuditCountMap() != null && auditWhitelistBlacklist.acceptTable(tableSpec.getTable().getDbName(), |
| tableSpec.getTable().getTableName())) { |
| tableMetadata.serializedAuditCountMaps.add(gmce.getAuditCountMap()); |
| } |
| if (gmce.getTopicPartitionOffsetsRange() != null) { |
| mergeOffsets(gmce, tid); |
| } |
| break; |
| } |
| case rewrite_files: { |
| updateTableProperty(tableSpec, tid); |
| rewriteFiles(gmce, newSpecsMap, oldSpecsMap, table, tableMetadata); |
| break; |
| } |
| case drop_files: { |
| dropFiles(gmce, oldSpecsMap, table, tableMetadata, tid); |
| break; |
| } |
| case change_property: { |
| updateTableProperty(tableSpec, tid); |
| if (gmce.getTopicPartitionOffsetsRange() != null) { |
| mergeOffsets(gmce, tid); |
| } |
| log.info("No file operation need to be performed by Iceberg Metadata Writer at this point."); |
| break; |
| } |
| default: { |
| log.error("unsupported operation {}", gmce.getOperationType().toString()); |
| return; |
| } |
| } |
| } |
| |
| private HashMap<String, List<Range>> getLastOffset(TableMetadata tableMetadata) { |
| HashMap<String, List<Range>> offsets = new HashMap<>(); |
| if (tableMetadata.lastProperties.isPresent()) { |
| for (Map.Entry<String, String> entry : tableMetadata.lastProperties.get().entrySet()) { |
| if (entry.getKey().startsWith(OFFSET_RANGE_KEY_PREFIX)) { |
| List<Range> ranges = Arrays.asList(entry.getValue().split(ConfigurationKeys.LIST_DELIMITER_KEY)) |
| .stream() |
| .map(s -> { |
| List<String> rangePair = Splitter.on(ConfigurationKeys.RANGE_DELIMITER_KEY).splitToList(s); |
| return Range.openClosed(Long.parseLong(rangePair.get(0)), Long.parseLong(rangePair.get(1)));}) |
| .collect(Collectors.toList()); |
| offsets.put(entry.getKey().substring(OFFSET_RANGE_KEY_PREFIX.length()), ranges); |
| } |
| } |
| } |
| return offsets; |
| } |
| |
| /** |
| * The side effect of this method is to update the offset-range of the table identified by |
| * the given {@link TableIdentifier} with the input {@link GobblinMetadataChangeEvent} |
| */ |
| private void mergeOffsets(GobblinMetadataChangeEvent gmce, TableIdentifier tid) { |
| TableMetadata tableMetadata = tableMetadataMap.computeIfAbsent(tid, t -> new TableMetadata()); |
| tableMetadata.dataOffsetRange = Optional.of(tableMetadata.dataOffsetRange.or(() -> getLastOffset(tableMetadata))); |
| Map<String, List<Range>> offsets = tableMetadata.dataOffsetRange.get(); |
| for (Map.Entry<String, String> entry : gmce.getTopicPartitionOffsetsRange().entrySet()) { |
| List<String> rangePair = Splitter.on(ConfigurationKeys.RANGE_DELIMITER_KEY).splitToList(entry.getValue()); |
| Range range = Range.openClosed(Long.parseLong(rangePair.get(0)), Long.parseLong(rangePair.get(1))); |
| if (range.lowerEndpoint().equals(range.upperEndpoint())) { |
| //Ignore this case |
| continue; |
| } |
| List<Range> existRanges = offsets.getOrDefault(entry.getKey(), new ArrayList<>()); |
| List<Range> newRanges = new ArrayList<>(); |
| for (Range r : existRanges) { |
| if (range.isConnected(r)) { |
| range = range.span(r); |
| } else { |
| newRanges.add(r); |
| } |
| } |
| newRanges.add(range); |
| Collections.sort(newRanges, new Comparator<Range>() { |
| @Override |
| public int compare(Range o1, Range o2) { |
| return o1.lowerEndpoint().compareTo(o2.lowerEndpoint()); |
| } |
| }); |
| offsets.put(entry.getKey(), newRanges); |
| } |
| } |
| |
| private void updateTableProperty(HiveSpec tableSpec, TableIdentifier tid) { |
| org.apache.hadoop.hive.metastore.api.Table table = HiveMetaStoreUtils.getTable(tableSpec.getTable()); |
| TableMetadata tableMetadata = tableMetadataMap.computeIfAbsent(tid, t -> new TableMetadata()); |
| tableMetadata.newProperties = Optional.of(IcebergUtils.getTableProperties(table)); |
| String nativeName = tableMetadata.datasetName; |
| String topic = nativeName.substring(nativeName.lastIndexOf("/") + 1); |
| tableMetadata.newProperties.get().put(TOPIC_NAME_KEY, topic); |
| } |
| |
| /** |
| * Compute the candidate schema from the gmce. |
| * If the schema source is schemaRegistry, we will use the schema creation time as the schema version to compute candidate schema and determine latest schema |
| * If the schema does not contain creation time, we will treat it the same as when schema source is event |
| * If the schema source is event, we will put it as default creation time, during flush, if we only have one candidate with default creation time, |
| * we'll use that to update schema. |
| * @param gmce |
| * @param tid |
| */ |
| private void computeCandidateSchema(GobblinMetadataChangeEvent gmce, TableIdentifier tid, HiveSpec spec) { |
| Table table = getIcebergTable(tid); |
| TableMetadata tableMetadata = tableMetadataMap.computeIfAbsent(tid, t -> new TableMetadata()); |
| org.apache.hadoop.hive.metastore.api.Table hiveTable = HiveMetaStoreUtils.getTable(spec.getTable()); |
| tableMetadata.lastProperties = Optional.of(tableMetadata.lastProperties.or(() -> table.properties())); |
| Map<String, String> props = tableMetadata.lastProperties.get(); |
| tableMetadata.lastSchemaVersion = Optional.of( |
| tableMetadata.lastSchemaVersion.or(() -> props.getOrDefault(SCHEMA_CREATION_TIME_KEY, DEFAULT_CREATION_TIME))); |
| String lastSchemaVersion = tableMetadata.lastSchemaVersion.get(); |
| tableMetadata.candidateSchemas = Optional.of(tableMetadata.candidateSchemas.or(() -> CacheBuilder.newBuilder() |
| .expireAfterAccess(conf.getInt(MetadataWriter.CACHE_EXPIRING_TIME, |
| MetadataWriter.DEFAULT_CACHE_EXPIRING_TIME), TimeUnit.HOURS) |
| .build())); |
| Cache<String, Schema> candidate = tableMetadata.candidateSchemas.get(); |
| try { |
| switch (gmce.getSchemaSource()) { |
| case SCHEMAREGISTRY: { |
| org.apache.avro.Schema schema = new org.apache.avro.Schema.Parser().parse(gmce.getTableSchema()); |
| String createdOn = AvroUtils.getSchemaCreationTime(schema); |
| if (createdOn == null) { |
| candidate.put(DEFAULT_CREATION_TIME, |
| IcebergUtils.getIcebergSchema(gmce.getTableSchema(), hiveTable).tableSchema); |
| } else if (!createdOn.equals(lastSchemaVersion)) { |
| candidate.put(createdOn, IcebergUtils.getIcebergSchema(gmce.getTableSchema(), hiveTable).tableSchema); |
| } |
| break; |
| } |
| case EVENT: { |
| candidate.put(DEFAULT_CREATION_TIME, |
| IcebergUtils.getIcebergSchema(gmce.getTableSchema(), hiveTable).tableSchema); |
| break; |
| } |
| case NONE: { |
| log.debug("Schema source set to be none, will ignore the schema"); |
| break; |
| } |
| default: { |
| throw new IOException(String.format("unsupported schema source %s", gmce.getSchemaSource())); |
| } |
| } |
| } catch (Exception e) { |
| log.error("Cannot get candidate schema from event due to", e); |
| } |
| } |
| |
| /** |
| * Add a partition column to the schema and partition spec |
| * @param table incoming iceberg table |
| * @param fieldName name of partition column |
| * @param type datatype of partition column |
| * @return table with updated schema and partition spec |
| */ |
| private Table addPartitionToIcebergTable(Table table, String fieldName, String type) { |
| if(!table.schema().columns().stream().anyMatch(x -> x.name().equalsIgnoreCase(fieldName))) { |
| table.updateSchema().addColumn(fieldName, Types.fromPrimitiveString(type)).commit(); |
| } |
| if(!table.spec().fields().stream().anyMatch(x -> x.name().equalsIgnoreCase(fieldName))) { |
| table.updateSpec().addField(fieldName).commit(); |
| } |
| table.refresh(); |
| return table; |
| } |
| |
| protected Table createTable(GobblinMetadataChangeEvent gmce, HiveSpec spec) throws IOException { |
| String schema = gmce.getTableSchema(); |
| org.apache.hadoop.hive.metastore.api.Table table = HiveMetaStoreUtils.getTable(spec.getTable()); |
| IcebergUtils.IcebergDataAndPartitionSchema schemas = IcebergUtils.getIcebergSchema(schema, table); |
| TableIdentifier tid = TableIdentifier.of(table.getDbName(), table.getTableName()); |
| Schema tableSchema = schemas.tableSchema; |
| Preconditions.checkState(tableSchema != null, "Table schema cannot be null when creating a table"); |
| PartitionSpec partitionSpec = IcebergUtils.getPartitionSpec(tableSchema, schemas.partitionSchema); |
| Table icebergTable = null; |
| String tableLocation = null; |
| if (useDataLocationAsTableLocation) { |
| tableLocation = gmce.getDatasetIdentifier().getNativeName() + String.format(TABLE_LOCATION_SUFFIX, table.getDbName()); |
| //Set the path permission |
| Path tablePath = new Path(tableLocation); |
| WriterUtils.mkdirsWithRecursivePermission(tablePath.getFileSystem(conf), tablePath, permission); |
| } |
| try (Timer.Context context = metricContext.timer(CREATE_TABLE_TIME).time()) { |
| icebergTable = |
| catalog.createTable(tid, tableSchema, partitionSpec, tableLocation, IcebergUtils.getTableProperties(table)); |
| log.info("Created table {}, schema: {} partition spec: {}", tid, tableSchema, partitionSpec); |
| } catch (AlreadyExistsException e) { |
| log.warn("table {} already exist, there may be some other process try to create table concurrently", tid); |
| } |
| return icebergTable; |
| } |
| |
| protected void rewriteFiles(GobblinMetadataChangeEvent gmce, Map<String, Collection<HiveSpec>> newSpecsMap, |
| Map<String, Collection<HiveSpec>> oldSpecsMap, Table table, TableMetadata tableMetadata) throws IOException { |
| PartitionSpec partitionSpec = table.spec(); |
| tableMetadata.ensureTxnInit(); |
| Set<DataFile> newDataFiles = new HashSet<>(); |
| getIcebergDataFilesToBeAddedHelper(gmce, table, newSpecsMap, tableMetadata) |
| .forEach(dataFile -> { |
| newDataFiles.add(dataFile); |
| tableMetadata.addedFiles.put(dataFile.path(), ""); |
| }); |
| Set<DataFile> oldDataFiles = getIcebergDataFilesToBeDeleted(gmce, table, newSpecsMap, oldSpecsMap, partitionSpec); |
| |
| // Dealing with the case when old file doesn't exist, in which it could either be converted into noop or AppendFile. |
| if (oldDataFiles.isEmpty() && !newDataFiles.isEmpty()) { |
| //We randomly check whether one of the new data files already exists in the db to avoid reprocessing re-write events |
| DataFile dataFile = newDataFiles.iterator().next(); |
| Expression exp = Expressions.startsWith(ICEBERG_FILE_PATH_COLUMN, dataFile.path().toString()); |
| if (FindFiles.in(table).withMetadataMatching(exp).collect().iterator().hasNext()) { |
| //This means this re-write event is duplicated with the one we already handled, so noop. |
| return; |
| } |
| // This is the case when the files to be deleted do not exist in table |
| // So we directly call addFiles interface to add new files into the table. |
| // Note that this AppendFiles won't be committed here, in contrast to a real rewrite operation |
| // where the commit will be called at once to save memory footprints. |
| AppendFiles appendFiles = tableMetadata.getOrInitAppendFiles(); |
| newDataFiles.forEach(appendFiles::appendFile); |
| return; |
| } |
| |
| tableMetadata.transaction.get().newRewrite().rewriteFiles(oldDataFiles, newDataFiles).commit(); |
| } |
| |
| /** |
| * Given the GMCE, get the iceberg schema with the origin ID specified by data pipeline which |
| * is corresponding to the file metrics index. |
| * @param gmce GMCE emitted by data pipeline |
| * @return iceberg schema with the origin ID |
| */ |
| private Schema getSchemaWithOriginId(GobblinMetadataChangeEvent gmce) { |
| Schema schemaWithOriginId = null; |
| if (gmce.getAvroSchemaWithIcebergSchemaID() != null) { |
| org.apache.iceberg.shaded.org.apache.avro.Schema avroSchema = |
| new org.apache.iceberg.shaded.org.apache.avro.Schema.Parser().parse(gmce.getAvroSchemaWithIcebergSchemaID()); |
| schemaWithOriginId = AvroSchemaUtil.toIceberg(avroSchema); |
| } |
| return schemaWithOriginId; |
| } |
| |
| /** |
| * Deal with both regular file deletions manifested by GMCE(aggregation but no commit), |
| * and expiring older snapshots(commit). |
| */ |
| protected void dropFiles(GobblinMetadataChangeEvent gmce, Map<String, Collection<HiveSpec>> oldSpecsMap, Table table, |
| TableMetadata tableMetadata, TableIdentifier tid) throws IOException { |
| PartitionSpec partitionSpec = table.spec(); |
| |
| // Update DeleteFiles in tableMetadata: This is regular file deletion |
| DeleteFiles deleteFiles = tableMetadata.getOrInitDeleteFiles(); |
| Set<DataFile> oldDataFiles = |
| getIcebergDataFilesToBeDeleted(gmce, table, new HashMap<>(), oldSpecsMap, partitionSpec); |
| oldDataFiles.forEach(deleteFiles::deleteFile); |
| |
| // Update ExpireSnapshots and commit the updates at once: This is for expiring snapshots that are |
| // beyond look-back allowance for time-travel. |
| parallelRunner.submitCallable(new Callable<Void>() { |
| @Override |
| public Void call() throws Exception { |
| try { |
| long olderThan = getExpireSnapshotTime(); |
| long start = System.currentTimeMillis(); |
| ExpireSnapshots expireSnapshots = table.expireSnapshots(); |
| final Table tmpTable = table; |
| |
| expireSnapshots.deleteWith(new Consumer<String>() { |
| @Override |
| public void accept(String file) { |
| if (file.startsWith(tmpTable.location())) { |
| tmpTable.io().deleteFile(file); |
| } |
| } |
| }).expireOlderThan(olderThan).commit(); |
| //TODO: emit these metrics to Ingraphs, in addition to metrics for publishing new snapshots and other Iceberg metadata operations. |
| log.info("Spent {} ms to expire snapshots older than {} ({}) in table {}", System.currentTimeMillis() - start, |
| new DateTime(olderThan).toString(), olderThan, tid.toString()); |
| } catch (Exception e) { |
| log.error(String.format("Fail to expire snapshots for table %s due to exception ", tid.toString()), e); |
| } |
| return null; |
| } |
| }, tid.toString()); |
| } |
| |
| private long getExpireSnapshotTime() { |
| PeriodFormatter periodFormatter = new PeriodFormatterBuilder().appendYears() |
| .appendSuffix("y") |
| .appendMonths() |
| .appendSuffix("M") |
| .appendDays() |
| .appendSuffix("d") |
| .appendHours() |
| .appendSuffix("h") |
| .appendMinutes() |
| .appendSuffix("m") |
| .toFormatter(); |
| return DateTime.now() |
| .minus(periodFormatter.parsePeriod( |
| conf.get(EXPIRE_SNAPSHOTS_LOOKBACK_TIME, DEFAULT_EXPIRE_SNAPSHOTS_LOOKBACK_TIME))) |
| .getMillis(); |
| } |
| |
| protected void addFiles(GobblinMetadataChangeEvent gmce, Map<String, Collection<HiveSpec>> newSpecsMap, Table table, |
| TableMetadata tableMetadata) { |
| AppendFiles appendFiles = tableMetadata.getOrInitAppendFiles(); |
| getIcebergDataFilesToBeAddedHelper(gmce, table, newSpecsMap, tableMetadata) |
| .forEach(dataFile -> { |
| appendFiles.appendFile(dataFile); |
| tableMetadata.addedFiles.put(dataFile.path(), ""); |
| }); |
| } |
| |
| private Stream<DataFile> getIcebergDataFilesToBeAddedHelper(GobblinMetadataChangeEvent gmce, Table table, |
| Map<String, Collection<HiveSpec>> newSpecsMap, |
| TableMetadata tableMetadata) { |
| return getIcebergDataFilesToBeAdded(table, tableMetadata, gmce, gmce.getNewFiles(), table.spec(), newSpecsMap, |
| IcebergUtils.getSchemaIdMap(getSchemaWithOriginId(gmce), table.schema())).stream() |
| .filter(dataFile -> tableMetadata.addedFiles.getIfPresent(dataFile.path()) == null); |
| } |
| |
| /** |
| * Method to get a {@link DataFile} collection without metrics information |
| * This method is used to get files to be deleted from iceberg |
| * If oldFilePrefixes is specified in gmce, this method will use those prefixes to find old file in iceberg, |
| * or the method will call method {IcebergUtils.getIcebergDataFileWithMetric} to get DataFile for specific file path |
| */ |
| private Set<DataFile> getIcebergDataFilesToBeDeleted(GobblinMetadataChangeEvent gmce, Table table, |
| Map<String, Collection<HiveSpec>> newSpecsMap, Map<String, Collection<HiveSpec>> oldSpecsMap, |
| PartitionSpec partitionSpec) throws IOException { |
| Set<DataFile> oldDataFiles = new HashSet<>(); |
| if (gmce.getOldFilePrefixes() != null) { |
| Expression exp = Expressions.alwaysFalse(); |
| for (String prefix : gmce.getOldFilePrefixes()) { |
| // Use both full path and raw path to filter old files |
| exp = Expressions.or(exp, Expressions.startsWith(ICEBERG_FILE_PATH_COLUMN, prefix)); |
| String rawPathPrefix = new Path(prefix).toUri().getRawPath(); |
| exp = Expressions.or(exp, Expressions.startsWith(ICEBERG_FILE_PATH_COLUMN, rawPathPrefix)); |
| } |
| long start = System.currentTimeMillis(); |
| oldDataFiles.addAll(Sets.newHashSet(FindFiles.in(table).withMetadataMatching(exp).collect().iterator())); |
| //Use INFO level log here to get better estimate. |
| //This shouldn't overwhelm the log since we receive limited number of rewrite_file gmces for one table in a day |
| log.info("Spent {}ms to query all old files in iceberg.", System.currentTimeMillis() - start); |
| } else { |
| for (String file : gmce.getOldFiles()) { |
| String specPath = new Path(file).getParent().toString(); |
| // For the use case of recompaction, the old path may contains /daily path, in this case, we find the spec from newSpecsMap |
| StructLike partitionVal = getIcebergPartitionVal( |
| oldSpecsMap.containsKey(specPath) ? oldSpecsMap.get(specPath) : newSpecsMap.get(specPath), file, |
| partitionSpec); |
| oldDataFiles.add(IcebergUtils.getIcebergDataFileWithoutMetric(file, partitionSpec, partitionVal)); |
| } |
| } |
| return oldDataFiles; |
| } |
| |
| /** |
| * Method to get dataFiles with metrics information |
| * This method is used to get files to be added to iceberg |
| * if completeness is enabled a new field (late) is added to table schema and partition spec |
| * computed based on datepartition and completion watermark |
| * This method will call method {IcebergUtils.getIcebergDataFileWithMetric} to get DataFile for specific file path |
| */ |
| private Set<DataFile> getIcebergDataFilesToBeAdded(Table table, TableMetadata tableMetadata, GobblinMetadataChangeEvent gmce, List<org.apache.gobblin.metadata.DataFile> files, |
| PartitionSpec partitionSpec, Map<String, Collection<HiveSpec>> newSpecsMap, Map<Integer, Integer> schemaIdMap) { |
| Set<DataFile> dataFiles = new HashSet<>(); |
| for (org.apache.gobblin.metadata.DataFile file : files) { |
| try { |
| Collection<HiveSpec> hiveSpecs = newSpecsMap.get(new Path(file.getFilePath()).getParent().toString()); |
| StructLike partition = getIcebergPartitionVal(hiveSpecs, file.getFilePath(), partitionSpec); |
| |
| if(tableMetadata.newPartitionColumnEnabled && gmce.getOperationType() == OperationType.add_files) { |
| // Assumes first partition value to be partitioned by date |
| // TODO Find better way to determine a partition value |
| String datepartition = partition.get(0, null); |
| partition = addLatePartitionValueToIcebergTable(table, tableMetadata, |
| hiveSpecs.iterator().next().getPartition().get(), datepartition); |
| tableMetadata.datePartitions.add(getDateTimeFromDatepartitionString(datepartition)); |
| } |
| dataFiles.add(IcebergUtils.getIcebergDataFileWithMetric(file, table.spec(), partition, conf, schemaIdMap)); |
| } catch (Exception e) { |
| log.warn("Cannot get DataFile for {} dur to {}", file.getFilePath(), e); |
| } |
| } |
| return dataFiles; |
| } |
| |
| /** |
| * 1. Add "late" partition column to iceberg table if not exists |
| * 2. compute "late" partition value based on datepartition and completion watermark |
| * 3. Default to late=0 if completion watermark check is disabled |
| * @param table |
| * @param tableMetadata |
| * @param hivePartition |
| * @param datepartition |
| * @return new iceberg partition value for file |
| */ |
| private StructLike addLatePartitionValueToIcebergTable(Table table, TableMetadata tableMetadata, HivePartition hivePartition, String datepartition) { |
| table = addPartitionToIcebergTable(table, newPartitionColumn, newPartitionColumnType); |
| PartitionSpec partitionSpec = table.spec(); |
| int late = !tableMetadata.completenessEnabled ? 0 : isLate(datepartition, tableMetadata.completionWatermark); |
| List<String> partitionValues = new ArrayList<>(hivePartition.getValues()); |
| partitionValues.add(String.valueOf(late)); |
| return IcebergUtils.getPartition(partitionSpec.partitionType(), partitionValues); |
| } |
| |
| private int isLate(String datepartition, long previousWatermark) { |
| ZonedDateTime partitionDateTime = ZonedDateTime.parse(datepartition, HOURLY_DATEPARTITION_FORMAT); |
| long partitionEpochTime = partitionDateTime.toInstant().toEpochMilli(); |
| if(partitionEpochTime >= previousWatermark) { |
| return 0; |
| } else if(partitionEpochTime < previousWatermark |
| && partitionDateTime.toLocalDate().equals(getDateFromEpochMillis(previousWatermark))) { |
| return 1; |
| } else { |
| return 2; |
| } |
| } |
| |
| private LocalDate getDateFromEpochMillis(long epochMillis) { |
| return ZonedDateTime.ofInstant(Instant.ofEpochMilli(epochMillis), ZoneId.of(timeZone)).toLocalDate(); |
| } |
| |
| private ZonedDateTime getDateTimeFromDatepartitionString(String datepartition) { |
| return ZonedDateTime.parse(datepartition, HOURLY_DATEPARTITION_FORMAT); |
| } |
| |
| /** |
| * Obtain Iceberg partition value with a collection of {@link HiveSpec}. |
| * @param specs A collection of {@link HiveSpec}s. |
| * @param filePath URI of file, used for logging purpose in this method. |
| * @param partitionSpec The scheme of partition. |
| * @return The value of partition based on the given {@link PartitionSpec}. |
| * @throws IOException |
| */ |
| private StructLike getIcebergPartitionVal(Collection<HiveSpec> specs, String filePath, PartitionSpec partitionSpec) |
| throws IOException { |
| if (specs == null || specs.isEmpty()) { |
| throw new IOException("Cannot get hive spec for " + filePath); |
| } |
| HivePartition hivePartition = specs.iterator().next().getPartition().orNull(); |
| StructLike partitionVal = hivePartition == null ? null |
| : IcebergUtils.getPartition(partitionSpec.partitionType(), hivePartition.getValues()); |
| return partitionVal; |
| } |
| |
| /** |
| * For flush of each table, we do the following logic: |
| * 1. Commit the appendFiles if it exist |
| * 2. Update the new table property: high watermark of GMCE, data offset range, schema versions |
| * 3. Update the schema |
| * 4. Commit the transaction |
| * 5. reset tableMetadata |
| * @param dbName |
| * @param tableName |
| */ |
| @Override |
| public void flush(String dbName, String tableName) throws IOException { |
| Lock writeLock = readWriteLock.writeLock(); |
| writeLock.lock(); |
| try { |
| TableIdentifier tid = TableIdentifier.of(dbName, tableName); |
| TableMetadata tableMetadata = tableMetadataMap.getOrDefault(tid, new TableMetadata()); |
| if (tableMetadata.transaction.isPresent()) { |
| Transaction transaction = tableMetadata.transaction.get(); |
| Map<String, String> props = tableMetadata.newProperties.or( |
| Maps.newHashMap(tableMetadata.lastProperties.or(getIcebergTable(tid).properties()))); |
| String topic = props.get(TOPIC_NAME_KEY); |
| if (tableMetadata.appendFiles.isPresent()) { |
| tableMetadata.appendFiles.get().commit(); |
| sendAuditCounts(topic, tableMetadata.serializedAuditCountMaps); |
| if (tableMetadata.completenessEnabled) { |
| checkAndUpdateCompletenessWatermark(tableMetadata, topic, tableMetadata.datePartitions, props); |
| } |
| } |
| if (tableMetadata.deleteFiles.isPresent()) { |
| tableMetadata.deleteFiles.get().commit(); |
| } |
| // Check and update completion watermark when there are no files to be registered, typically for quiet topics |
| // The logic is to check the next window from previous completion watermark and update the watermark if there are no audit counts |
| if(!tableMetadata.appendFiles.isPresent() && !tableMetadata.deleteFiles.isPresent() |
| && tableMetadata.completenessEnabled) { |
| if (tableMetadata.completionWatermark > DEFAULT_COMPLETION_WATERMARK) { |
| log.info(String.format("Checking kafka audit for %s on change_property ", topic)); |
| SortedSet<ZonedDateTime> timestamps = new TreeSet<>(); |
| ZonedDateTime prevWatermarkDT = |
| Instant.ofEpochMilli(tableMetadata.completionWatermark).atZone(ZoneId.of(this.timeZone)); |
| timestamps.add(TimeIterator.inc(prevWatermarkDT, TimeIterator.Granularity.valueOf(this.auditCheckGranularity), 1)); |
| checkAndUpdateCompletenessWatermark(tableMetadata, topic, timestamps, props); |
| } else { |
| log.info(String.format("Need valid watermark, current watermark is %s, Not checking kafka audit for %s", |
| tableMetadata.completionWatermark, topic)); |
| } |
| } |
| |
| //Set high waterMark |
| Long highWatermark = tableCurrentWatermarkMap.get(tid); |
| props.put(String.format(GMCE_HIGH_WATERMARK_KEY, tableTopicPartitionMap.get(tid)), highWatermark.toString()); |
| //Set low waterMark |
| props.put(String.format(GMCE_LOW_WATERMARK_KEY, tableTopicPartitionMap.get(tid)), |
| tableMetadata.lowWatermark.get().toString()); |
| //Set whether to delete metadata files after commit |
| props.put(TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED, Boolean.toString( |
| conf.getBoolean(TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED, |
| TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED_DEFAULT))); |
| props.put(TableProperties.METADATA_PREVIOUS_VERSIONS_MAX, Integer.toString( |
| conf.getInt(TableProperties.METADATA_PREVIOUS_VERSIONS_MAX, |
| TableProperties.METADATA_PREVIOUS_VERSIONS_MAX_DEFAULT))); |
| //Set data offset range |
| boolean containOffsetRange = setDatasetOffsetRange(tableMetadata, props); |
| String topicName = tableName; |
| if (containOffsetRange) { |
| String topicPartitionString = tableMetadata.dataOffsetRange.get().keySet().iterator().next(); |
| //In case the topic name is not the table name or the topic name contains '-' |
| topicName = topicPartitionString.substring(0, topicPartitionString.lastIndexOf('-')); |
| } |
| //Update schema(commit) |
| updateSchema(tableMetadata, props, topicName); |
| //Update properties |
| UpdateProperties updateProperties = transaction.updateProperties(); |
| props.forEach(updateProperties::set); |
| updateProperties.commit(); |
| try (AutoCloseableHiveLock lock = this.locks.getTableLock(dbName, tableName)) { |
| transaction.commitTransaction(); |
| } |
| |
| // Emit GTE for snapshot commits |
| Snapshot snapshot = tableMetadata.table.get().currentSnapshot(); |
| Map<String, String> currentProps = tableMetadata.table.get().properties(); |
| submitSnapshotCommitEvent(snapshot, tableMetadata, dbName, tableName, currentProps, highWatermark); |
| |
| //Reset the table metadata for next accumulation period |
| tableMetadata.reset(currentProps, highWatermark); |
| log.info(String.format("Finish commit of new snapshot %s for table %s", snapshot.snapshotId(), tid.toString())); |
| } else { |
| log.info("There's no transaction initiated for the table {}", tid.toString()); |
| } |
| } catch (RuntimeException e) { |
| throw new RuntimeException(String.format("Fail to flush table %s %s", dbName, tableName), e); |
| } catch (Exception e) { |
| throw new IOException(String.format("Fail to flush table %s %s", dbName, tableName), e); |
| } finally { |
| writeLock.unlock(); |
| } |
| } |
| |
| @Override |
| public void reset(String dbName, String tableName) throws IOException { |
| this.tableMetadataMap.remove(TableIdentifier.of(dbName, tableName)); |
| } |
| |
| /** |
| * Update TableMetadata with the new completion watermark upon a successful audit check |
| * @param tableMetadata metadata of table |
| * @param topic topic name |
| * @param timestamps Sorted set in reverse order of timestamps to check audit counts for |
| * @param props table properties map |
| */ |
| private void checkAndUpdateCompletenessWatermark(TableMetadata tableMetadata, String topic, SortedSet<ZonedDateTime> timestamps, |
| Map<String, String> props) { |
| if (topic == null) { |
| log.error(String.format("Not performing audit check. %s is null. Please set as table property of %s", |
| TOPIC_NAME_KEY, tableMetadata.table.get().name())); |
| } |
| long newCompletenessWatermark = |
| computeCompletenessWatermark(topic, timestamps, tableMetadata.completionWatermark); |
| if (newCompletenessWatermark > tableMetadata.completionWatermark) { |
| log.info(String.format("Updating %s for %s to %s", COMPLETION_WATERMARK_KEY, tableMetadata.table.get().name(), |
| newCompletenessWatermark)); |
| props.put(COMPLETION_WATERMARK_KEY, String.valueOf(newCompletenessWatermark)); |
| props.put(COMPLETION_WATERMARK_TIMEZONE_KEY, this.timeZone); |
| tableMetadata.completionWatermark = newCompletenessWatermark; |
| } |
| } |
| |
| /** |
| * NOTE: completion watermark for a window [t1, t2] is marked as t2 if audit counts match |
| * for that window (aka its is set to the beginning of next window) |
| * For each timestamp in sorted collection of timestamps in descending order |
| * if timestamp is greater than previousWatermark |
| * and hour(now) > hour(prevWatermark) |
| * check audit counts for completeness between |
| * a source and reference tier for [timestamp -1 , timstamp unit of granularity] |
| * If the audit count matches update the watermark to the timestamp and break |
| * else continue |
| * else |
| * break |
| * Using a {@link TimeIterator} that operates over a range of time in 1 unit |
| * given the start, end and granularity |
| * @param table |
| * @param timestamps a sorted set of timestamps in decreasing order |
| * @param previousWatermark previous completion watermark for the table |
| * @return updated completion watermark |
| */ |
| private long computeCompletenessWatermark(String table, SortedSet<ZonedDateTime> timestamps, long previousWatermark) { |
| log.info(String.format("Compute completion watermark for %s and timestamps %s with previous watermark %s", table, timestamps, previousWatermark)); |
| long completionWatermark = previousWatermark; |
| ZonedDateTime now = ZonedDateTime.now(ZoneId.of(this.timeZone)); |
| try { |
| if(timestamps == null || timestamps.size() <= 0) { |
| log.error("Cannot create time iterator. Empty for null timestamps"); |
| return previousWatermark; |
| } |
| TimeIterator.Granularity granularity = TimeIterator.Granularity.valueOf(this.auditCheckGranularity); |
| ZonedDateTime prevWatermarkDT = Instant.ofEpochMilli(previousWatermark) |
| .atZone(ZoneId.of(this.timeZone)); |
| ZonedDateTime startDT = timestamps.first(); |
| ZonedDateTime endDT = timestamps.last(); |
| TimeIterator iterator = new TimeIterator(startDT, endDT, granularity, true); |
| while (iterator.hasNext()) { |
| ZonedDateTime timestampDT = iterator.next(); |
| if (timestampDT.isAfter(prevWatermarkDT) |
| && TimeIterator.durationBetween(prevWatermarkDT, now, granularity) > 0) { |
| long timestampMillis = timestampDT.toInstant().toEpochMilli(); |
| if (auditCountVerifier.get().isComplete(table, |
| TimeIterator.dec(timestampDT, granularity, 1).toInstant().toEpochMilli(), timestampMillis)) { |
| completionWatermark = timestampMillis; |
| break; |
| } |
| } else { |
| break; |
| } |
| } |
| } catch (IOException e) { |
| log.warn("Exception during audit count check: ", e); |
| } |
| return completionWatermark; |
| } |
| |
| private void submitSnapshotCommitEvent(Snapshot snapshot, TableMetadata tableMetadata, String dbName, |
| String tableName, Map<String, String> props, Long highWaterMark) { |
| GobblinEventBuilder gobblinTrackingEvent = |
| new GobblinEventBuilder(MetadataWriterKeys.ICEBERG_COMMIT_EVENT_NAME); |
| long currentSnapshotID = snapshot.snapshotId(); |
| long endToEndLag = System.currentTimeMillis() - tableMetadata.lowestGMCEEmittedTime; |
| TableIdentifier tid = TableIdentifier.of(dbName, tableName); |
| String gmceTopicPartition = tableTopicPartitionMap.get(tid); |
| |
| //Add information to automatically trigger repair jon when data loss happen |
| gobblinTrackingEvent.addMetadata(MetadataWriterKeys.GMCE_TOPIC_NAME, gmceTopicPartition.split("-")[0]); |
| gobblinTrackingEvent.addMetadata(MetadataWriterKeys.GMCE_TOPIC_PARTITION, gmceTopicPartition.split("-")[1]); |
| gobblinTrackingEvent.addMetadata(MetadataWriterKeys.GMCE_HIGH_WATERMARK, highWaterMark.toString()); |
| gobblinTrackingEvent.addMetadata(MetadataWriterKeys.GMCE_LOW_WATERMARK, |
| tableMetadata.lowWatermark.get().toString()); |
| |
| //Add information for lag monitoring |
| gobblinTrackingEvent.addMetadata(MetadataWriterKeys.LAG_KEY_NAME, Long.toString(endToEndLag)); |
| gobblinTrackingEvent.addMetadata(MetadataWriterKeys.SNAPSHOT_KEY_NAME, Long.toString(currentSnapshotID)); |
| gobblinTrackingEvent.addMetadata(MetadataWriterKeys.MANIFEST_LOCATION, snapshot.manifestListLocation()); |
| gobblinTrackingEvent.addMetadata(MetadataWriterKeys.SNAPSHOT_INFORMATION_KEY_NAME, |
| Joiner.on(",").withKeyValueSeparator("=").join(snapshot.summary())); |
| gobblinTrackingEvent.addMetadata(MetadataWriterKeys.ICEBERG_TABLE_KEY_NAME, tableName); |
| gobblinTrackingEvent.addMetadata(MetadataWriterKeys.ICEBERG_DATABASE_KEY_NAME, dbName); |
| gobblinTrackingEvent.addMetadata(MetadataWriterKeys.DATASET_HDFS_PATH, tableMetadata.datasetName); |
| for (Map.Entry<String, String> entry : props.entrySet()) { |
| if (entry.getKey().startsWith(OFFSET_RANGE_KEY_PREFIX)) { |
| gobblinTrackingEvent.addMetadata(entry.getKey(), entry.getValue()); |
| } |
| } |
| eventSubmitter.submit(gobblinTrackingEvent); |
| } |
| |
| private boolean setDatasetOffsetRange(TableMetadata tableMetadata, Map<String, String> props) { |
| if (tableMetadata.dataOffsetRange.isPresent() && !tableMetadata.dataOffsetRange.get().isEmpty()) { |
| for (Map.Entry<String, List<Range>> offsets : tableMetadata.dataOffsetRange.get().entrySet()) { |
| List<Range> ranges = offsets.getValue(); |
| String rangeString = ranges.stream() |
| .map(r -> Joiner.on(ConfigurationKeys.RANGE_DELIMITER_KEY).join(r.lowerEndpoint(), r.upperEndpoint())) |
| .collect(Collectors.joining(ConfigurationKeys.LIST_DELIMITER_KEY)); |
| props.put(String.format(OFFSET_RANGE_KEY_FORMAT, offsets.getKey()), rangeString); |
| } |
| return true; |
| } |
| return false; |
| } |
| |
| private void updateSchema(TableMetadata tableMetadata, Map<String, String> props, String topicName) { |
| //Set default schema versions |
| props.put(SCHEMA_CREATION_TIME_KEY, tableMetadata.lastSchemaVersion.or(DEFAULT_CREATION_TIME)); |
| // Update Schema |
| try { |
| if (tableMetadata.candidateSchemas.isPresent() && tableMetadata.candidateSchemas.get().size() > 0) { |
| Cache candidates = tableMetadata.candidateSchemas.get(); |
| //Only have default schema, so either we calculate schema from event or the schema does not have creation time, directly update it |
| if (candidates.size() == 1 && candidates.getIfPresent(DEFAULT_CREATION_TIME) != null) { |
| updateSchemaHelper(DEFAULT_CREATION_TIME, (Schema) candidates.getIfPresent(DEFAULT_CREATION_TIME), props, |
| tableMetadata.table.get()); |
| } else { |
| //update schema if candidates contains the schema that has the same creation time with the latest schema |
| org.apache.avro.Schema latestSchema = |
| (org.apache.avro.Schema) schemaRegistry.getLatestSchemaByTopic(topicName); |
| String creationTime = AvroUtils.getSchemaCreationTime(latestSchema); |
| if (creationTime == null) { |
| log.warn( |
| "Schema from schema registry does not contain creation time, check config for schema registry class"); |
| } else if (candidates.getIfPresent(creationTime) != null) { |
| updateSchemaHelper(creationTime, (Schema) candidates.getIfPresent(creationTime), props, |
| tableMetadata.table.get()); |
| } |
| } |
| } |
| } catch (SchemaRegistryException e) { |
| log.error("Cannot get schema form schema registry, will not update this schema", e); |
| } |
| } |
| |
| private void updateSchemaHelper(String schemaCreationTime, Schema schema, Map<String, String> props, Table table) { |
| try { |
| table.updateSchema().unionByNameWith(schema).commit(); |
| props.put(SCHEMA_CREATION_TIME_KEY, schemaCreationTime); |
| } catch (Exception e) { |
| log.error("Cannot update schema to " + schema.toString() + "for table " + table.location(), e); |
| } |
| } |
| |
| @Override |
| public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope, Map<String, Collection<HiveSpec>> newSpecsMap, |
| Map<String, Collection<HiveSpec>> oldSpecsMap, HiveSpec tableSpec) throws IOException { |
| Lock readLock = readWriteLock.readLock(); |
| readLock.lock(); |
| try { |
| GenericRecord genericRecord = recordEnvelope.getRecord(); |
| GobblinMetadataChangeEvent gmce = |
| (GobblinMetadataChangeEvent) SpecificData.get().deepCopy(genericRecord.getSchema(), genericRecord); |
| String dbName = tableSpec.getTable().getDbName(); |
| String tableName = tableSpec.getTable().getTableName(); |
| if (whitelistBlacklist.acceptTable(dbName, tableName)) { |
| TableIdentifier tid = TableIdentifier.of(dbName, tableName); |
| String topicPartition = tableTopicPartitionMap.computeIfAbsent(tid, |
| t -> recordEnvelope.getWatermark().getSource()); |
| Long currentWatermark = getAndPersistCurrentWatermark(tid, topicPartition); |
| Long currentOffset = ((LongWatermark)recordEnvelope.getWatermark().getWatermark()).getValue(); |
| |
| if (currentOffset > currentWatermark) { |
| if (!tableMetadataMap.computeIfAbsent(tid, t -> new TableMetadata()).lowWatermark.isPresent()) { |
| //This means we haven't register this table or met some error before, we need to reset the low watermark |
| tableMetadataMap.get(tid).lowWatermark = Optional.of(currentOffset - 1); |
| tableMetadataMap.get(tid).setDatasetName(gmce.getDatasetIdentifier().getNativeName()); |
| if (this.newPartitionEnabled && this.newPartitionTableWhitelistBlacklist.acceptTable(dbName, tableName)) { |
| tableMetadataMap.get(tid).newPartitionColumnEnabled = true; |
| if (this.completenessEnabled && this.completenessWhitelistBlacklist.acceptTable(dbName, tableName)) { |
| tableMetadataMap.get(tid).completenessEnabled = true; |
| } |
| } |
| } |
| |
| write(gmce, newSpecsMap, oldSpecsMap, tableSpec); |
| tableCurrentWatermarkMap.put(tid, currentOffset); |
| } else { |
| log.warn(String.format("Skip processing record %s since it has lower watermark", genericRecord.toString())); |
| } |
| } else { |
| log.info(String.format("Skip table %s.%s since it's not selected", tableSpec.getTable().getDbName(), |
| tableSpec.getTable().getTableName())); |
| } |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| /** |
| * Method to send audit counts given a topic name and a list of serialized audit count maps. Called only when new files |
| * are added. Default is no-op, must be implemented in a subclass. |
| */ |
| public void sendAuditCounts(String topicName, Collection<String> serializedAuditCountMaps) { |
| } |
| |
| @Override |
| public void close() throws IOException { |
| this.closer.close(); |
| } |
| |
| /** |
| * A collection of Iceberg metadata including {@link Table} itself as well as |
| * A set of buffered objects (reflecting table's {@link org.apache.iceberg.PendingUpdate}s) within the flush interval |
| * that aggregates the metadata like location arriving / deleting files, schema, |
| * along with other table-level metadata like watermark, offset-range, etc. |
| * |
| * Also note the difference with {@link org.apache.iceberg.TableMetadata}. |
| */ |
| private class TableMetadata { |
| Optional<Table> table = Optional.absent(); |
| |
| /** |
| * The {@link Transaction} object holds the reference of a {@link org.apache.iceberg.BaseTransaction.TransactionTableOperations} |
| * that is shared by all individual operation (e.g. {@link AppendFiles}) to ensure atomicity even if commit method |
| * is invoked from a individual operation. |
| */ |
| Optional<Transaction> transaction = Optional.absent(); |
| private Optional<AppendFiles> appendFiles = Optional.absent(); |
| private Optional<DeleteFiles> deleteFiles = Optional.absent(); |
| |
| Optional<Map<String, String>> lastProperties = Optional.absent(); |
| Optional<Map<String, String>> newProperties = Optional.absent(); |
| Optional<Cache<String, Schema>> candidateSchemas = Optional.absent(); |
| Optional<Map<String, List<Range>>> dataOffsetRange = Optional.absent(); |
| Optional<String> lastSchemaVersion = Optional.absent(); |
| Optional<Long> lowWatermark = Optional.absent(); |
| long completionWatermark = DEFAULT_COMPLETION_WATERMARK; |
| SortedSet<ZonedDateTime> datePartitions = new TreeSet<>(Collections.reverseOrder()); |
| List<String> serializedAuditCountMaps = new ArrayList<>(); |
| |
| @Setter |
| String datasetName; |
| boolean completenessEnabled; |
| boolean newPartitionColumnEnabled; |
| |
| Cache<CharSequence, String> addedFiles = CacheBuilder.newBuilder() |
| .expireAfterAccess(conf.getInt(ADDED_FILES_CACHE_EXPIRING_TIME, DEFAULT_ADDED_FILES_CACHE_EXPIRING_TIME), |
| TimeUnit.HOURS) |
| .build(); |
| long lowestGMCEEmittedTime = Long.MAX_VALUE; |
| |
| /** |
| * Always use this method to obtain {@link AppendFiles} object within flush interval |
| * if clients want to have the {@link AppendFiles} committed along with other updates in a txn. |
| */ |
| AppendFiles getOrInitAppendFiles() { |
| ensureTxnInit(); |
| if (!this.appendFiles.isPresent()) { |
| this.appendFiles = Optional.of(this.transaction.get().newAppend()); |
| } |
| |
| return this.appendFiles.get(); |
| } |
| |
| DeleteFiles getOrInitDeleteFiles() { |
| ensureTxnInit(); |
| if (!this.deleteFiles.isPresent()) { |
| this.deleteFiles = Optional.of(this.transaction.get().newDelete()); |
| } |
| |
| return this.deleteFiles.get(); |
| } |
| |
| /** |
| * Initializing {@link Transaction} object within {@link TableMetadata} when needed. |
| */ |
| void ensureTxnInit() { |
| if (!this.transaction.isPresent()) { |
| this.transaction = Optional.of(table.get().newTransaction()); |
| } |
| } |
| |
| void reset(Map<String, String> props, Long lowWaterMark) { |
| this.lastProperties = Optional.of(props); |
| this.lastSchemaVersion = Optional.of(props.get(SCHEMA_CREATION_TIME_KEY)); |
| this.transaction = Optional.absent(); |
| this.deleteFiles = Optional.absent(); |
| this.appendFiles = Optional.absent(); |
| |
| // Clean cache and reset to eagerly release unreferenced objects. |
| if (this.candidateSchemas.isPresent()) { |
| this.candidateSchemas.get().cleanUp(); |
| } |
| this.candidateSchemas = Optional.absent(); |
| |
| this.dataOffsetRange = Optional.absent(); |
| this.newProperties = Optional.absent(); |
| this.lowestGMCEEmittedTime = Long.MAX_VALUE; |
| this.lowWatermark = Optional.of(lowWaterMark); |
| this.datePartitions.clear(); |
| this.serializedAuditCountMaps.clear(); |
| } |
| } |
| } |
