You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by sj...@apache.org on 2015/08/25 20:08:52 UTC
[23/50] [abbrv] hadoop git commit: YARN-3411. [Storage
implementation] explore the native HBase write schema for storage (Vrushali C
via sjlee)
YARN-3411. [Storage implementation] explore the native HBase write schema for storage (Vrushali C via sjlee)
(cherry picked from commit 7a3068854d27eadae1c57545988f5b2029bf119a)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/b3edd630
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/b3edd630
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/b3edd630
Branch: refs/heads/YARN-2928
Commit: b3edd6301b4ec8c59928cca0d5a41c94fb23740d
Parents: e93fa60
Author: Sangjin Lee <sj...@apache.org>
Authored: Thu May 21 14:11:01 2015 -0700
Committer: Sangjin Lee <sj...@apache.org>
Committed: Tue Aug 25 10:47:12 2015 -0700
----------------------------------------------------------------------
hadoop-yarn-project/CHANGES.txt | 2 +
.../collector/TimelineCollectorManager.java | 19 +
.../storage/EntityColumnDetails.java | 110 ++++++
.../storage/EntityColumnFamily.java | 95 +++++
.../storage/HBaseTimelineWriterImpl.java | 225 ++++++++++++
.../server/timelineservice/storage/Range.java | 59 ++++
.../storage/TimelineEntitySchemaConstants.java | 71 ++++
.../storage/TimelineSchemaCreator.java | 231 +++++++++++++
.../storage/TimelineWriterUtils.java | 344 +++++++++++++++++++
.../storage/TestHBaseTimelineWriterImpl.java | 292 ++++++++++++++++
10 files changed, 1448 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b3edd630/hadoop-yarn-project/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/CHANGES.txt b/hadoop-yarn-project/CHANGES.txt
index ec9abc9..3df01e9 100644
--- a/hadoop-yarn-project/CHANGES.txt
+++ b/hadoop-yarn-project/CHANGES.txt
@@ -73,6 +73,8 @@ Branch YARN-2928: Timeline Server Next Generation: Phase 1
YARN-3634. TestMRTimelineEventHandling and TestApplication are broken. (
Sangjin Lee via junping_du)
+ YARN-3411. [Storage implementation] explore the native HBase write schema
+ for storage (Vrushali C via sjlee)
IMPROVEMENTS
OPTIMIZATIONS
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b3edd630/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/TimelineCollectorManager.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/TimelineCollectorManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/TimelineCollectorManager.java
index 953d9b7..d54715c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/TimelineCollectorManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/TimelineCollectorManager.java
@@ -59,6 +59,13 @@ public abstract class TimelineCollectorManager extends AbstractService {
super.serviceInit(conf);
}
+ @Override
+ protected void serviceStart() throws Exception {
+ super.serviceStart();
+ if (writer != null) {
+ writer.start();
+ }
+ }
// access to this map is synchronized with the map itself
private final Map<ApplicationId, TimelineCollector> collectors =
@@ -147,4 +154,16 @@ public abstract class TimelineCollectorManager extends AbstractService {
return collectors.containsKey(appId);
}
+ @Override
+ protected void serviceStop() throws Exception {
+ if (collectors != null && collectors.size() > 1) {
+ for (TimelineCollector c : collectors.values()) {
+ c.serviceStop();
+ }
+ }
+ if (writer != null) {
+ writer.close();
+ }
+ super.serviceStop();
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b3edd630/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/EntityColumnDetails.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/EntityColumnDetails.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/EntityColumnDetails.java
new file mode 100644
index 0000000..2894c41
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/EntityColumnDetails.java
@@ -0,0 +1,110 @@
+/**
+ * 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.hadoop.yarn.server.timelineservice.storage;
+
+import java.io.IOException;
+import java.util.Set;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hbase.client.BufferedMutator;
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * Contains the Info Column Family details like Column names, types and byte
+ * representations for
+ * {@link org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity}
+ * object that is stored in hbase Also has utility functions for storing each of
+ * these to the backend
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+enum EntityColumnDetails {
+ ID(EntityColumnFamily.INFO, "id"),
+ TYPE(EntityColumnFamily.INFO, "type"),
+ CREATED_TIME(EntityColumnFamily.INFO, "created_time"),
+ MODIFIED_TIME(EntityColumnFamily.INFO, "modified_time"),
+ FLOW_VERSION(EntityColumnFamily.INFO, "flow_version"),
+ PREFIX_IS_RELATED_TO(EntityColumnFamily.INFO, "r"),
+ PREFIX_RELATES_TO(EntityColumnFamily.INFO, "s"),
+ PREFIX_EVENTS(EntityColumnFamily.INFO, "e");
+
+ private final EntityColumnFamily columnFamily;
+ private final String value;
+ private final byte[] inBytes;
+
+ private EntityColumnDetails(EntityColumnFamily columnFamily,
+ String value) {
+ this.columnFamily = columnFamily;
+ this.value = value;
+ this.inBytes = Bytes.toBytes(this.value.toLowerCase());
+ }
+
+ public String getValue() {
+ return value;
+ }
+
+ byte[] getInBytes() {
+ return inBytes;
+ }
+
+ void store(byte[] rowKey, BufferedMutator entityTable, Object inputValue)
+ throws IOException {
+ TimelineWriterUtils.store(rowKey, entityTable,
+ this.columnFamily.getInBytes(), null, this.getInBytes(), inputValue,
+ null);
+ }
+
+ /**
+ * stores events data with column prefix
+ */
+ void store(byte[] rowKey, BufferedMutator entityTable, byte[] idBytes,
+ String key, Object inputValue) throws IOException {
+ TimelineWriterUtils.store(rowKey, entityTable,
+ this.columnFamily.getInBytes(),
+ // column prefix
+ TimelineWriterUtils.join(
+ TimelineEntitySchemaConstants.ROW_KEY_SEPARATOR_BYTES,
+ this.getInBytes(), idBytes),
+ // column qualifier
+ Bytes.toBytes(key),
+ inputValue, null);
+ }
+
+ /**
+ * stores relation entities with a column prefix
+ */
+ void store(byte[] rowKey, BufferedMutator entityTable, String key,
+ Set<String> inputValue) throws IOException {
+ TimelineWriterUtils.store(rowKey, entityTable,
+ this.columnFamily.getInBytes(),
+ // column prefix
+ this.getInBytes(),
+ // column qualifier
+ Bytes.toBytes(key),
+ // value
+ TimelineWriterUtils.getValueAsString(
+ TimelineEntitySchemaConstants.ROW_KEY_SEPARATOR, inputValue),
+ // cell timestamp
+ null);
+ }
+
+ // TODO add a method that accepts a byte array,
+ // iterates over the enum and returns an enum from those bytes
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b3edd630/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/EntityColumnFamily.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/EntityColumnFamily.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/EntityColumnFamily.java
new file mode 100644
index 0000000..e556351
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/EntityColumnFamily.java
@@ -0,0 +1,95 @@
+/**
+ * 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.hadoop.yarn.server.timelineservice.storage;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hbase.client.BufferedMutator;
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * Contains the Column family names and byte representations for
+ * {@link org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity}
+ * object that is stored in hbase
+ * Also has utility functions for storing each of these to the backend
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+enum EntityColumnFamily {
+ INFO("i"),
+ CONFIG("c"),
+ METRICS("m");
+
+ private final String value;
+ private final byte[] inBytes;
+
+ private EntityColumnFamily(String value) {
+ this.value = value;
+ this.inBytes = Bytes.toBytes(this.value.toLowerCase());
+ }
+
+ byte[] getInBytes() {
+ return inBytes;
+ }
+
+ public String getValue() {
+ return value;
+ }
+
+ /**
+ * stores the key as column and value as hbase column value in the given
+ * column family in the entity table
+ *
+ * @param rowKey
+ * @param entityTable
+ * @param inputValue
+ * @throws IOException
+ */
+ public void store(byte[] rowKey, BufferedMutator entityTable, String key,
+ String inputValue) throws IOException {
+ if (key == null) {
+ return;
+ }
+ TimelineWriterUtils.store(rowKey, entityTable, this.getInBytes(), null,
+ Bytes.toBytes(key), inputValue, null);
+ }
+
+ /**
+ * stores the values along with cell timestamp
+ *
+ * @param rowKey
+ * @param entityTable
+ * @param key
+ * @param timestamp
+ * @param inputValue
+ * @throws IOException
+ */
+ public void store(byte[] rowKey, BufferedMutator entityTable, String key,
+ Long timestamp, Number inputValue) throws IOException {
+ if (key == null) {
+ return;
+ }
+ TimelineWriterUtils.store(rowKey, entityTable, this.getInBytes(), null,
+ Bytes.toBytes(key), inputValue, timestamp);
+ }
+
+ // TODO add a method that accepts a byte array,
+ // iterates over the enum and returns an enum from those bytes
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b3edd630/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/HBaseTimelineWriterImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/HBaseTimelineWriterImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/HBaseTimelineWriterImpl.java
new file mode 100644
index 0000000..aa71c6c
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/HBaseTimelineWriterImpl.java
@@ -0,0 +1,225 @@
+/**
+ * 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.hadoop.yarn.server.timelineservice.storage;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.service.AbstractService;
+import org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntities;
+import org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity;
+import org.apache.hadoop.yarn.api.records.timelineservice.TimelineEvent;
+import org.apache.hadoop.yarn.api.records.timelineservice.TimelineMetric;
+import org.apache.hadoop.yarn.api.records.timelineservice.TimelineWriteResponse;
+import org.apache.hadoop.hbase.client.BufferedMutator;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.yarn.server.timelineservice.storage.TimelineEntitySchemaConstants;
+
+/**
+ * This implements a hbase based backend for storing application timeline entity
+ * information.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public class HBaseTimelineWriterImpl extends AbstractService implements
+ TimelineWriter {
+
+ private Connection conn;
+ private BufferedMutator entityTable;
+
+ private static final Log LOG = LogFactory
+ .getLog(HBaseTimelineWriterImpl.class);
+
+ public HBaseTimelineWriterImpl() {
+ super(HBaseTimelineWriterImpl.class.getName());
+ }
+
+ public HBaseTimelineWriterImpl(Configuration conf) throws IOException {
+ super(conf.get("yarn.application.id",
+ HBaseTimelineWriterImpl.class.getName()));
+ }
+
+ /**
+ * initializes the hbase connection to write to the entity table
+ */
+ @Override
+ protected void serviceInit(Configuration conf) throws Exception {
+ super.serviceInit(conf);
+ Configuration hbaseConf = HBaseConfiguration.create(conf);
+ conn = ConnectionFactory.createConnection(hbaseConf);
+ TableName entityTableName = TableName.valueOf(hbaseConf.get(
+ TimelineEntitySchemaConstants.ENTITY_TABLE_NAME,
+ TimelineEntitySchemaConstants.DEFAULT_ENTITY_TABLE_NAME));
+ entityTable = conn.getBufferedMutator(entityTableName);
+ }
+
+ /**
+ * Stores the entire information in TimelineEntities to the timeline store.
+ */
+ @Override
+ public TimelineWriteResponse write(String clusterId, String userId,
+ String flowName, String flowVersion, long flowRunId, String appId,
+ TimelineEntities data) throws IOException {
+
+ byte[] rowKeyPrefix = TimelineWriterUtils.getRowKeyPrefix(clusterId,
+ userId, flowName, flowRunId, appId);
+
+ TimelineWriteResponse putStatus = new TimelineWriteResponse();
+ for (TimelineEntity te : data.getEntities()) {
+
+ // a set can have at most 1 null
+ if (te == null) {
+ continue;
+ }
+ // get row key
+ byte[] row = TimelineWriterUtils.join(
+ TimelineEntitySchemaConstants.ROW_KEY_SEPARATOR_BYTES, rowKeyPrefix,
+ Bytes.toBytes(te.getType()), Bytes.toBytes(te.getId()));
+
+ storeInfo(row, te, flowVersion);
+ storeEvents(row, te.getEvents());
+ storeConfig(row, te.getConfigs());
+ storeMetrics(row, te.getMetrics());
+ storeRelations(row, te.getIsRelatedToEntities(),
+ EntityColumnDetails.PREFIX_IS_RELATED_TO);
+ storeRelations(row, te.getRelatesToEntities(),
+ EntityColumnDetails.PREFIX_RELATES_TO);
+ }
+
+ return putStatus;
+ }
+
+ /**
+ * Stores the Relations from the {@linkplain TimelineEntity} object
+ */
+ private void storeRelations(byte[] rowKey,
+ Map<String, Set<String>> connectedEntities,
+ EntityColumnDetails columnNamePrefix) throws IOException {
+ for (Map.Entry<String, Set<String>> entry : connectedEntities.entrySet()) {
+ columnNamePrefix.store(rowKey, entityTable, entry.getKey(),
+ entry.getValue());
+ }
+ }
+
+ /**
+ * Stores information from the {@linkplain TimelineEntity} object
+ */
+ private void storeInfo(byte[] rowKey, TimelineEntity te, String flowVersion)
+ throws IOException {
+
+ EntityColumnDetails.ID.store(rowKey, entityTable, te.getId());
+ EntityColumnDetails.TYPE.store(rowKey, entityTable, te.getType());
+ EntityColumnDetails.CREATED_TIME.store(rowKey, entityTable,
+ te.getCreatedTime());
+ EntityColumnDetails.MODIFIED_TIME.store(rowKey, entityTable,
+ te.getModifiedTime());
+ EntityColumnDetails.FLOW_VERSION.store(rowKey, entityTable, flowVersion);
+ }
+
+ /**
+ * stores the config information from {@linkplain TimelineEntity}
+ */
+ private void storeConfig(byte[] rowKey, Map<String, String> config)
+ throws IOException {
+ if (config == null) {
+ return;
+ }
+ for (Map.Entry<String, String> entry : config.entrySet()) {
+ EntityColumnFamily.CONFIG.store(rowKey, entityTable,
+ entry.getKey(), entry.getValue());
+ }
+ }
+
+ /**
+ * stores the {@linkplain TimelineMetric} information from the
+ * {@linkplain TimelineEvent} object
+ */
+ private void storeMetrics(byte[] rowKey, Set<TimelineMetric> metrics)
+ throws IOException {
+ if (metrics != null) {
+ for (TimelineMetric metric : metrics) {
+ String key = metric.getId();
+ Map<Long, Number> timeseries = metric.getValues();
+ for (Map.Entry<Long, Number> entry : timeseries.entrySet()) {
+ EntityColumnFamily.METRICS.store(rowKey, entityTable, key,
+ entry.getKey(), entry.getValue());
+ }
+ }
+ }
+ }
+
+ /**
+ * Stores the events from the {@linkplain TimelineEvent} object
+ */
+ private void storeEvents(byte[] rowKey, Set<TimelineEvent> events)
+ throws IOException {
+ if (events != null) {
+ for (TimelineEvent event : events) {
+ if (event != null) {
+ String id = event.getId();
+ if (id != null) {
+ byte[] idBytes = Bytes.toBytes(id);
+ Map<String, Object> eventInfo = event.getInfo();
+ if (eventInfo != null) {
+ for (Map.Entry<String, Object> info : eventInfo.entrySet()) {
+ EntityColumnDetails.PREFIX_EVENTS.store(rowKey,
+ entityTable, idBytes, info.getKey(), info.getValue());
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public TimelineWriteResponse aggregate(TimelineEntity data,
+ TimelineAggregationTrack track) throws IOException {
+ return null;
+ }
+
+ /**
+ * close the hbase connections
+ * The close APIs perform flushing and release any
+ * resources held
+ */
+ @Override
+ protected void serviceStop() throws Exception {
+ if (entityTable != null) {
+ LOG.info("closing entity table");
+ // The close API performs flushing and releases any resources held
+ entityTable.close();
+ }
+ if (conn != null) {
+ LOG.info("closing the hbase Connection");
+ conn.close();
+ }
+ super.serviceStop();
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b3edd630/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/Range.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/Range.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/Range.java
new file mode 100644
index 0000000..2a2db81
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/Range.java
@@ -0,0 +1,59 @@
+/**
+ * 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.hadoop.yarn.server.timelineservice.storage;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public class Range {
+ private final int startIdx;
+ private final int endIdx;
+
+ /**
+ * Defines a range from start index (inclusive) to end index (exclusive).
+ *
+ * @param start
+ * Starting index position
+ * @param end
+ * Ending index position (exclusive)
+ */
+ public Range(int start, int end) {
+ if (start < 0 || end < start) {
+ throw new IllegalArgumentException(
+ "Invalid range, required that: 0 <= start <= end; start=" + start
+ + ", end=" + end);
+ }
+
+ this.startIdx = start;
+ this.endIdx = end;
+ }
+
+ public int start() {
+ return startIdx;
+ }
+
+ public int end() {
+ return endIdx;
+ }
+
+ public int length() {
+ return endIdx - startIdx;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b3edd630/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/TimelineEntitySchemaConstants.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/TimelineEntitySchemaConstants.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/TimelineEntitySchemaConstants.java
new file mode 100644
index 0000000..d95cbb2
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/TimelineEntitySchemaConstants.java
@@ -0,0 +1,71 @@
+/**
+ * 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.hadoop.yarn.server.timelineservice.storage;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+
+/**
+ * contains the constants used in the context of schema accesses for
+ * {@link org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity}
+ * information
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public class TimelineEntitySchemaConstants {
+
+ /** entity prefix */
+ public static final String ENTITY_PREFIX =
+ YarnConfiguration.TIMELINE_SERVICE_PREFIX
+ + ".entity";
+
+ /** config param name that specifies the entity table name */
+ public static final String ENTITY_TABLE_NAME = ENTITY_PREFIX
+ + ".table.name";
+
+ /**
+ * config param name that specifies the TTL for metrics column family in
+ * entity table
+ */
+ public static final String ENTITY_TABLE_METRICS_TTL = ENTITY_PREFIX
+ + ".table.metrics.ttl";
+
+ /** default value for entity table name */
+ public static final String DEFAULT_ENTITY_TABLE_NAME = "timelineservice.entity";
+
+ /** in bytes default value for entity table name */
+ static final byte[] DEFAULT_ENTITY_TABLE_NAME_BYTES = Bytes
+ .toBytes(DEFAULT_ENTITY_TABLE_NAME);
+
+ /** separator in row key */
+ public static final String ROW_KEY_SEPARATOR = "!";
+
+ /** byte representation of the separator in row key */
+ static final byte[] ROW_KEY_SEPARATOR_BYTES = Bytes
+ .toBytes(ROW_KEY_SEPARATOR);
+
+ public static final byte ZERO_BYTES = 0;
+
+ /** default TTL is 30 days for metrics timeseries */
+ public static final int ENTITY_TABLE_METRICS_TTL_DEFAULT = 2592000;
+
+ /** default max number of versions */
+ public static final int ENTITY_TABLE_METRICS_MAX_VERSIONS_DEFAULT = 1000;
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b3edd630/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/TimelineSchemaCreator.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/TimelineSchemaCreator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/TimelineSchemaCreator.java
new file mode 100644
index 0000000..820a6d1
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/TimelineSchemaCreator.java
@@ -0,0 +1,231 @@
+/**
+ * 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.hadoop.yarn.server.timelineservice.storage;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.regionserver.BloomType;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.commons.cli.PosixParser;
+import org.apache.hadoop.util.GenericOptionsParser;
+
+/**
+ * This creates the schema for a hbase based backend for storing application
+ * timeline information.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public class TimelineSchemaCreator {
+
+ final static String NAME = TimelineSchemaCreator.class.getSimpleName();
+ private static final Log LOG = LogFactory.getLog(TimelineSchemaCreator.class);
+ final static byte[][] splits = { Bytes.toBytes("a"), Bytes.toBytes("ad"),
+ Bytes.toBytes("an"), Bytes.toBytes("b"), Bytes.toBytes("ca"),
+ Bytes.toBytes("cl"), Bytes.toBytes("d"), Bytes.toBytes("e"),
+ Bytes.toBytes("f"), Bytes.toBytes("g"), Bytes.toBytes("h"),
+ Bytes.toBytes("i"), Bytes.toBytes("j"), Bytes.toBytes("k"),
+ Bytes.toBytes("l"), Bytes.toBytes("m"), Bytes.toBytes("n"),
+ Bytes.toBytes("o"), Bytes.toBytes("q"), Bytes.toBytes("r"),
+ Bytes.toBytes("s"), Bytes.toBytes("se"), Bytes.toBytes("t"),
+ Bytes.toBytes("u"), Bytes.toBytes("v"), Bytes.toBytes("w"),
+ Bytes.toBytes("x"), Bytes.toBytes("y"), Bytes.toBytes("z") };
+
+ public static final String SPLIT_KEY_PREFIX_LENGTH = "4";
+
+ public static void main(String[] args) throws Exception {
+
+ Configuration hbaseConf = HBaseConfiguration.create();
+ // Grab input args and allow for -Dxyz style arguments
+ String[] otherArgs = new GenericOptionsParser(hbaseConf, args)
+ .getRemainingArgs();
+
+ // Grab the arguments we're looking for.
+ CommandLine commandLine = parseArgs(otherArgs);
+
+ // Grab the entityTableName argument
+ String entityTableName = commandLine.getOptionValue("e");
+ if (StringUtils.isNotBlank(entityTableName)) {
+ hbaseConf.set(TimelineEntitySchemaConstants.ENTITY_TABLE_NAME,
+ entityTableName);
+ }
+ String entityTable_TTL_Metrics = commandLine.getOptionValue("m");
+ if (StringUtils.isNotBlank(entityTable_TTL_Metrics)) {
+ hbaseConf.set(TimelineEntitySchemaConstants.ENTITY_TABLE_METRICS_TTL,
+ entityTable_TTL_Metrics);
+ }
+ createAllTables(hbaseConf);
+ }
+
+ /**
+ * Parse command-line arguments.
+ *
+ * @param args
+ * command line arguments passed to program.
+ * @return parsed command line.
+ * @throws ParseException
+ */
+ private static CommandLine parseArgs(String[] args) throws ParseException {
+ Options options = new Options();
+
+ // Input
+ Option o = new Option("e", "entityTableName", true, "entity table name");
+ o.setArgName("entityTableName");
+ o.setRequired(false);
+ options.addOption(o);
+
+ o = new Option("m", "metricsTTL", true, "TTL for metrics column family");
+ o.setArgName("metricsTTL");
+ o.setRequired(false);
+ options.addOption(o);
+
+ CommandLineParser parser = new PosixParser();
+ CommandLine commandLine = null;
+ try {
+ commandLine = parser.parse(options, args);
+ } catch (Exception e) {
+ LOG.error("ERROR: " + e.getMessage() + "\n");
+ HelpFormatter formatter = new HelpFormatter();
+ formatter.printHelp(NAME + " ", options, true);
+ System.exit(-1);
+ }
+
+ return commandLine;
+ }
+
+ private static void createAllTables(Configuration hbaseConf)
+ throws IOException {
+
+ Connection conn = null;
+ try {
+ conn = ConnectionFactory.createConnection(hbaseConf);
+ Admin admin = conn.getAdmin();
+ if (admin == null) {
+ throw new IOException("Cannot create table since admin is null");
+ }
+ createTimelineEntityTable(admin, hbaseConf);
+ } finally {
+ if (conn != null) {
+ conn.close();
+ }
+ }
+ }
+
+ /**
+ * Creates a table with column families info, config and metrics
+ * info stores information about a timeline entity object
+ * config stores configuration data of a timeline entity object
+ * metrics stores the metrics of a timeline entity object
+ *
+ * Example entity table record:
+ * <pre>
+ *|------------------------------------------------------------|
+ *| Row | Column Family | Column Family | Column Family|
+ *| key | info | metrics | config |
+ *|------------------------------------------------------------|
+ *| userName! | id:entityId | metricName1: | configKey1: |
+ *| clusterId! | | metricValue1 | configValue1 |
+ *| flowId! | type:entityType| @timestamp1 | |
+ *| flowRunId! | | | configKey2: |
+ *| AppId! | created_time: | metricName1: | configValue2 |
+ *| entityType!| 1392993084018 | metricValue2 | |
+ *| entityId | | @timestamp2 | |
+ *| | modified_time: | | |
+ *| | 1392995081012 | metricName2: | |
+ *| | | metricValue1 | |
+ *| | r!relatesToKey:| @timestamp2 | |
+ *| | id3!id4!id5 | | |
+ *| | | | |
+ *| | s!isRelatedToKey| | |
+ *| | id7!id9!id5 | | |
+ *| | | | |
+ *| | e!eventKey: | | |
+ *| | eventValue | | |
+ *| | | | |
+ *| | flowVersion: | | |
+ *| | versionValue | | |
+ *|------------------------------------------------------------|
+ *</pre>
+ * @param admin
+ * @param hbaseConf
+ * @throws IOException
+ */
+ public static void createTimelineEntityTable(Admin admin,
+ Configuration hbaseConf) throws IOException {
+
+ TableName table = TableName.valueOf(hbaseConf.get(
+ TimelineEntitySchemaConstants.ENTITY_TABLE_NAME,
+ TimelineEntitySchemaConstants.DEFAULT_ENTITY_TABLE_NAME));
+ if (admin.tableExists(table)) {
+ // do not disable / delete existing table
+ // similar to the approach taken by map-reduce jobs when
+ // output directory exists
+ throw new IOException("Table " + table.getNameAsString()
+ + " already exists.");
+ }
+
+ HTableDescriptor entityTableDescp = new HTableDescriptor(table);
+ HColumnDescriptor cf1 = new HColumnDescriptor(
+ EntityColumnFamily.INFO.getInBytes());
+ cf1.setBloomFilterType(BloomType.ROWCOL);
+ entityTableDescp.addFamily(cf1);
+
+ HColumnDescriptor cf2 = new HColumnDescriptor(
+ EntityColumnFamily.CONFIG.getInBytes());
+ cf2.setBloomFilterType(BloomType.ROWCOL);
+ cf2.setBlockCacheEnabled(true);
+ entityTableDescp.addFamily(cf2);
+
+ HColumnDescriptor cf3 = new HColumnDescriptor(
+ EntityColumnFamily.METRICS.getInBytes());
+ entityTableDescp.addFamily(cf3);
+ cf3.setBlockCacheEnabled(true);
+ // always keep 1 version (the latest)
+ cf3.setMinVersions(1);
+ cf3.setMaxVersions(TimelineEntitySchemaConstants.ENTITY_TABLE_METRICS_MAX_VERSIONS_DEFAULT);
+ cf3.setTimeToLive(hbaseConf.getInt(
+ TimelineEntitySchemaConstants.ENTITY_TABLE_METRICS_TTL,
+ TimelineEntitySchemaConstants.ENTITY_TABLE_METRICS_TTL_DEFAULT));
+ entityTableDescp
+ .setRegionSplitPolicyClassName("org.apache.hadoop.hbase.regionserver.KeyPrefixRegionSplitPolicy");
+ entityTableDescp.setValue("KeyPrefixRegionSplitPolicy.prefix_length",
+ SPLIT_KEY_PREFIX_LENGTH);
+ admin.createTable(entityTableDescp, splits);
+ LOG.info("Status of table creation for " + table.getNameAsString() + "="
+ + admin.tableExists(table));
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b3edd630/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/TimelineWriterUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/TimelineWriterUtils.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/TimelineWriterUtils.java
new file mode 100644
index 0000000..113935e
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/TimelineWriterUtils.java
@@ -0,0 +1,344 @@
+/**
+ * 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.hadoop.yarn.server.timelineservice.storage;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.client.BufferedMutator;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.yarn.server.timeline.GenericObjectMapper;
+import org.apache.hadoop.yarn.server.timelineservice.storage.Range;
+
+/**
+ * bunch of utility functions used across TimelineWriter classes
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public class TimelineWriterUtils {
+
+ /** empty bytes */
+ public static final byte[] EMPTY_BYTES = new byte[0];
+ private static final String SPACE = " ";
+ private static final String UNDERSCORE = "_";
+ private static final String EMPTY_STRING = "";
+
+ /**
+ * Returns a single byte array containing all of the individual component
+ * arrays separated by the separator array.
+ *
+ * @param separator
+ * @param components
+ * @return byte array after joining the components
+ */
+ public static byte[] join(byte[] separator, byte[]... components) {
+ if (components == null || components.length == 0) {
+ return EMPTY_BYTES;
+ }
+
+ int finalSize = 0;
+ if (separator != null) {
+ finalSize = separator.length * (components.length - 1);
+ }
+ for (byte[] comp : components) {
+ if (comp != null) {
+ finalSize += comp.length;
+ }
+ }
+
+ byte[] buf = new byte[finalSize];
+ int offset = 0;
+ for (int i = 0; i < components.length; i++) {
+ if (components[i] != null) {
+ System.arraycopy(components[i], 0, buf, offset, components[i].length);
+ offset += components[i].length;
+ if (i < (components.length - 1) && separator != null
+ && separator.length > 0) {
+ System.arraycopy(separator, 0, buf, offset, separator.length);
+ offset += separator.length;
+ }
+ }
+ }
+ return buf;
+ }
+
+ /**
+ * Splits the source array into multiple array segments using the given
+ * separator, up to a maximum of count items. This will naturally produce
+ * copied byte arrays for each of the split segments. To identify the split
+ * ranges without the array copies, see
+ * {@link TimelineWriterUtils#splitRanges(byte[], byte[])}.
+ *
+ * @param source
+ * @param separator
+ * @return byte[] array after splitting the source
+ */
+ public static byte[][] split(byte[] source, byte[] separator) {
+ return split(source, separator, -1);
+ }
+
+ /**
+ * Splits the source array into multiple array segments using the given
+ * separator, up to a maximum of count items. This will naturally produce
+ * copied byte arrays for each of the split segments. To identify the split
+ * ranges without the array copies, see
+ * {@link TimelineWriterUtils#splitRanges(byte[], byte[])}.
+ *
+ * @param source
+ * @param separator
+ * @param limit
+ * @return byte[][] after splitting the input source
+ */
+ public static byte[][] split(byte[] source, byte[] separator, int limit) {
+ List<Range> segments = splitRanges(source, separator, limit);
+
+ byte[][] splits = new byte[segments.size()][];
+ for (int i = 0; i < segments.size(); i++) {
+ Range r = segments.get(i);
+ byte[] tmp = new byte[r.length()];
+ if (tmp.length > 0) {
+ System.arraycopy(source, r.start(), tmp, 0, r.length());
+ }
+ splits[i] = tmp;
+ }
+ return splits;
+ }
+
+ /**
+ * Returns a list of ranges identifying [start, end) -- closed, open --
+ * positions within the source byte array that would be split using the
+ * separator byte array.
+ */
+ public static List<Range> splitRanges(byte[] source, byte[] separator) {
+ return splitRanges(source, separator, -1);
+ }
+
+ /**
+ * Returns a list of ranges identifying [start, end) -- closed, open --
+ * positions within the source byte array that would be split using the
+ * separator byte array.
+ * @param source the source data
+ * @param separator the separator pattern to look for
+ * @param limit the maximum number of splits to identify in the source
+ */
+ public static List<Range> splitRanges(byte[] source, byte[] separator, int limit) {
+ List<Range> segments = new ArrayList<Range>();
+ if ((source == null) || (separator == null)) {
+ return segments;
+ }
+ int start = 0;
+ itersource: for (int i = 0; i < source.length; i++) {
+ for (int j = 0; j < separator.length; j++) {
+ if (source[i + j] != separator[j]) {
+ continue itersource;
+ }
+ }
+ // all separator elements matched
+ if (limit > 0 && segments.size() >= (limit-1)) {
+ // everything else goes in one final segment
+ break;
+ }
+
+ segments.add(new Range(start, i));
+ start = i + separator.length;
+ // i will be incremented again in outer for loop
+ i += separator.length-1;
+ }
+ // add in remaining to a final range
+ if (start <= source.length) {
+ segments.add(new Range(start, source.length));
+ }
+ return segments;
+ }
+
+ /**
+ * converts run id into it's inverse timestamp
+ * @param flowRunId
+ * @return inverted long
+ */
+ public static long encodeRunId(Long flowRunId) {
+ return Long.MAX_VALUE - flowRunId;
+ }
+
+ /**
+ * return a value from the Map as a String
+ * @param key
+ * @param values
+ * @return value as a String or ""
+ * @throws IOException
+ */
+ public static String getValueAsString(final byte[] key,
+ final Map<byte[], byte[]> values) throws IOException {
+ if( values == null ) {
+ return EMPTY_STRING;
+ }
+ byte[] value = values.get(key);
+ if (value != null) {
+ return GenericObjectMapper.read(value).toString();
+ } else {
+ return EMPTY_STRING;
+ }
+ }
+
+ /**
+ * return a value from the Map as a long
+ * @param key
+ * @param values
+ * @return value as Long or 0L
+ * @throws IOException
+ */
+ public static long getValueAsLong(final byte[] key,
+ final Map<byte[], byte[]> values) throws IOException {
+ if (values == null) {
+ return 0;
+ }
+ byte[] value = values.get(key);
+ if (value != null) {
+ Number val = (Number) GenericObjectMapper.read(value);
+ return val.longValue();
+ } else {
+ return 0L;
+ }
+ }
+
+ /**
+ * concates the values from a Set<Strings> to return a single delimited string value
+ * @param rowKeySeparator
+ * @param values
+ * @return Value from the set of strings as a string
+ */
+ public static String getValueAsString(String rowKeySeparator,
+ Set<String> values) {
+
+ if (values == null) {
+ return EMPTY_STRING;
+ }
+ StringBuilder concatStrings = new StringBuilder();
+ for (String value : values) {
+ concatStrings.append(value);
+ concatStrings.append(rowKeySeparator);
+ }
+ // remove the last separator
+ if(concatStrings.length() > 1) {
+ concatStrings.deleteCharAt(concatStrings.lastIndexOf(rowKeySeparator));
+ }
+ return concatStrings.toString();
+ }
+ /**
+ * Constructs a row key prefix for the entity table
+ * @param clusterId
+ * @param userId
+ * @param flowId
+ * @param flowRunId
+ * @param appId
+ * @return byte array with the row key prefix
+ */
+ static byte[] getRowKeyPrefix(String clusterId, String userId, String flowId,
+ Long flowRunId, String appId) {
+ return TimelineWriterUtils.join(
+ TimelineEntitySchemaConstants.ROW_KEY_SEPARATOR_BYTES,
+ Bytes.toBytes(cleanse(userId)), Bytes.toBytes(cleanse(clusterId)),
+ Bytes.toBytes(cleanse(flowId)),
+ Bytes.toBytes(TimelineWriterUtils.encodeRunId(flowRunId)),
+ Bytes.toBytes(cleanse(appId)));
+ }
+
+ /**
+ * Takes a string token to be used as a key or qualifier and
+ * cleanses out reserved tokens.
+ * This operation is not symmetrical.
+ * Logic is to replace all spaces and separator chars in input with
+ * underscores.
+ *
+ * @param token token to cleanse.
+ * @return String with no spaces and no separator chars
+ */
+ public static String cleanse(String token) {
+ if (token == null || token.length() == 0) {
+ return token;
+ }
+
+ String cleansed = token.replaceAll(SPACE, UNDERSCORE);
+ cleansed = cleansed.replaceAll(
+ TimelineEntitySchemaConstants.ROW_KEY_SEPARATOR, UNDERSCORE);
+
+ return cleansed;
+ }
+
+ /**
+ * stores the info to the table in hbase
+ *
+ * @param rowKey
+ * @param table
+ * @param columnFamily
+ * @param columnPrefix
+ * @param columnQualifier
+ * @param inputValue
+ * @param cellTimeStamp
+ * @throws IOException
+ */
+ public static void store(byte[] rowKey, BufferedMutator table, byte[] columnFamily,
+ byte[] columnPrefix, byte[] columnQualifier, Object inputValue,
+ Long cellTimeStamp) throws IOException {
+ if ((rowKey == null) || (table == null) || (columnFamily == null)
+ || (columnQualifier == null) || (inputValue == null)) {
+ return;
+ }
+
+ Put p = null;
+ if (cellTimeStamp == null) {
+ if (columnPrefix != null) {
+ // store with prefix
+ p = new Put(rowKey);
+ p.addColumn(
+ columnFamily,
+ join(TimelineEntitySchemaConstants.ROW_KEY_SEPARATOR_BYTES,
+ columnPrefix, columnQualifier), GenericObjectMapper
+ .write(inputValue));
+ } else {
+ // store without prefix
+ p = new Put(rowKey);
+ p.addColumn(columnFamily, columnQualifier,
+ GenericObjectMapper.write(inputValue));
+ }
+ } else {
+ // store with cell timestamp
+ Cell cell = CellUtil.createCell(rowKey, columnFamily, columnQualifier,
+ // set the cell timestamp
+ cellTimeStamp,
+ // KeyValue Type minimum
+ TimelineEntitySchemaConstants.ZERO_BYTES,
+ GenericObjectMapper.write(inputValue));
+ p = new Put(rowKey);
+ p.add(cell);
+ }
+ if (p != null) {
+ table.mutate(p);
+ }
+
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b3edd630/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/test/java/org/apache/hadoop/yarn/server/timelineservice/storage/TestHBaseTimelineWriterImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/test/java/org/apache/hadoop/yarn/server/timelineservice/storage/TestHBaseTimelineWriterImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/test/java/org/apache/hadoop/yarn/server/timelineservice/storage/TestHBaseTimelineWriterImpl.java
new file mode 100644
index 0000000..48bacd6
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/test/java/org/apache/hadoop/yarn/server/timelineservice/storage/TestHBaseTimelineWriterImpl.java
@@ -0,0 +1,292 @@
+/**
+ * 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.hadoop.yarn.server.timelineservice.storage;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Set;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertNotNull;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntities;
+import org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity;
+import org.apache.hadoop.yarn.api.records.timelineservice.TimelineMetric;
+import org.apache.hadoop.yarn.api.records.timelineservice.TimelineMetric.Type;
+import org.apache.hadoop.yarn.server.timeline.GenericObjectMapper;
+import org.junit.BeforeClass;
+import org.junit.AfterClass;
+import org.junit.Test;
+
+/**
+ * Unit test HBaseTimelineWriterImpl
+ * YARN 3411
+ *
+ * @throws Exception
+ */
+public class TestHBaseTimelineWriterImpl {
+
+ private static HBaseTestingUtility util;
+
+ @BeforeClass
+ public static void setupBeforeClass() throws Exception {
+ util = new HBaseTestingUtility();
+ util.startMiniCluster();
+ createSchema();
+ }
+
+ private static void createSchema() throws IOException {
+ byte[][] families = new byte[3][];
+ families[0] = EntityColumnFamily.INFO.getInBytes();
+ families[1] = EntityColumnFamily.CONFIG.getInBytes();
+ families[2] = EntityColumnFamily.METRICS.getInBytes();
+ TimelineSchemaCreator.createTimelineEntityTable(util.getHBaseAdmin(),
+ util.getConfiguration());
+ }
+
+ @Test
+ public void testWriteEntityToHBase() throws Exception {
+ TimelineEntities te = new TimelineEntities();
+ TimelineEntity entity = new TimelineEntity();
+ String id = "hello";
+ String type = "world";
+ entity.setId(id);
+ entity.setType(type);
+ Long cTime = 1425016501000L;
+ Long mTime = 1425026901000L;
+ entity.setCreatedTime(cTime);
+ entity.setModifiedTime(mTime);
+
+ // add the isRelatedToEntity info
+ String key = "task";
+ String value = "is_related_to_entity_id_here";
+ Set<String> isRelatedToSet = new HashSet<String>();
+ isRelatedToSet.add(value);
+ Map<String, Set<String>> isRelatedTo = new HashMap<String, Set<String>>();
+ isRelatedTo.put(key, isRelatedToSet);
+ entity.setIsRelatedToEntities(isRelatedTo);
+
+ // add the relatesTo info
+ key = "container";
+ value = "relates_to_entity_id_here";
+ Set<String> relatesToSet = new HashSet<String>();
+ relatesToSet.add(value);
+ value = "relates_to_entity_id_here_Second";
+ relatesToSet.add(value);
+ Map<String, Set<String>> relatesTo = new HashMap<String, Set<String>>();
+ relatesTo.put(key, relatesToSet);
+ entity.setRelatesToEntities(relatesTo);
+
+ // add some config entries
+ Map<String, String> conf = new HashMap<String, String>();
+ conf.put("config_param1", "value1");
+ conf.put("config_param2", "value2");
+ entity.addConfigs(conf);
+
+ // add metrics
+ Set<TimelineMetric> metrics = new HashSet<>();
+ TimelineMetric m1 = new TimelineMetric();
+ m1.setId("MAP_SLOT_MILLIS");
+ Map<Long, Number> metricValues = new HashMap<Long, Number>();
+ metricValues.put(1429741609000L, 100000000);
+ metricValues.put(1429742609000L, 200000000);
+ metricValues.put(1429743609000L, 300000000);
+ metricValues.put(1429744609000L, 400000000);
+ metricValues.put(1429745609000L, 50000000000L);
+ metricValues.put(1429746609000L, 60000000000L);
+ m1.setType(Type.TIME_SERIES);
+ m1.setValues(metricValues);
+ metrics.add(m1);
+ entity.addMetrics(metrics);
+
+ te.addEntity(entity);
+
+ HBaseTimelineWriterImpl hbi = null;
+ try {
+ Configuration c1 = util.getConfiguration();
+ hbi = new HBaseTimelineWriterImpl(c1);
+ hbi.init(c1);
+ String cluster = "cluster1";
+ String user = "user1";
+ String flow = "some_flow_name";
+ String flowVersion = "AB7822C10F1111";
+ long runid = 1002345678919L;
+ String appName = "some app name";
+ hbi.write(cluster, user, flow, flowVersion, runid, appName, te);
+ hbi.stop();
+
+ // scan the table and see that entity exists
+ Scan s = new Scan();
+ byte[] startRow = TimelineWriterUtils.getRowKeyPrefix(cluster, user, flow,
+ runid, appName);
+ s.setStartRow(startRow);
+ s.setMaxVersions(Integer.MAX_VALUE);
+ ResultScanner scanner = null;
+ TableName entityTableName = TableName
+ .valueOf(TimelineEntitySchemaConstants.DEFAULT_ENTITY_TABLE_NAME);
+ Connection conn = ConnectionFactory.createConnection(c1);
+ Table entityTable = conn.getTable(entityTableName);
+ int rowCount = 0;
+ int colCount = 0;
+ scanner = entityTable.getScanner(s);
+ for (Result result : scanner) {
+ if (result != null && !result.isEmpty()) {
+ rowCount++;
+ colCount += result.size();
+ byte[] row1 = result.getRow();
+ assertTrue(isRowKeyCorrect(row1, cluster, user, flow, runid, appName,
+ entity));
+
+ // check info column family
+ NavigableMap<byte[], byte[]> infoValues = result
+ .getFamilyMap(EntityColumnFamily.INFO.getInBytes());
+ String id1 = TimelineWriterUtils.getValueAsString(
+ EntityColumnDetails.ID.getInBytes(), infoValues);
+ assertEquals(id, id1);
+ String type1 = TimelineWriterUtils.getValueAsString(
+ EntityColumnDetails.TYPE.getInBytes(), infoValues);
+ assertEquals(type, type1);
+ Long cTime1 = TimelineWriterUtils.getValueAsLong(
+ EntityColumnDetails.CREATED_TIME.getInBytes(), infoValues);
+ assertEquals(cTime1, cTime);
+ Long mTime1 = TimelineWriterUtils.getValueAsLong(
+ EntityColumnDetails.MODIFIED_TIME.getInBytes(), infoValues);
+ assertEquals(mTime1, mTime);
+ checkRelatedEntities(isRelatedTo, infoValues,
+ EntityColumnDetails.PREFIX_IS_RELATED_TO.getInBytes());
+ checkRelatedEntities(relatesTo, infoValues,
+ EntityColumnDetails.PREFIX_RELATES_TO.getInBytes());
+
+ // check config column family
+ NavigableMap<byte[], byte[]> configValuesResult = result
+ .getFamilyMap(EntityColumnFamily.CONFIG.getInBytes());
+ checkConfigs(configValuesResult, conf);
+
+ NavigableMap<byte[], byte[]> metricsResult = result
+ .getFamilyMap(EntityColumnFamily.METRICS.getInBytes());
+ checkMetricsSizeAndKey(metricsResult, metrics);
+ List<Cell> metricCells = result.getColumnCells(
+ EntityColumnFamily.METRICS.getInBytes(),
+ Bytes.toBytes(m1.getId()));
+ checkMetricsTimeseries(metricCells, m1);
+ }
+ }
+ assertEquals(1, rowCount);
+ assertEquals(15, colCount);
+
+ } finally {
+ hbi.stop();
+ hbi.close();
+ }
+ }
+
+ private void checkMetricsTimeseries(List<Cell> metricCells,
+ TimelineMetric m1) throws IOException {
+ Map<Long, Number> timeseries = m1.getValues();
+ assertEquals(metricCells.size(), timeseries.size());
+ for (Cell c1 : metricCells) {
+ assertTrue(timeseries.containsKey(c1.getTimestamp()));
+ assertEquals(GenericObjectMapper.read(CellUtil.cloneValue(c1)),
+ timeseries.get(c1.getTimestamp()));
+ }
+ }
+
+ private void checkMetricsSizeAndKey(
+ NavigableMap<byte[], byte[]> metricsResult, Set<TimelineMetric> metrics) {
+ assertEquals(metrics.size(), metricsResult.size());
+ for (TimelineMetric m1 : metrics) {
+ byte[] key = Bytes.toBytes(m1.getId());
+ assertTrue(metricsResult.containsKey(key));
+ }
+ }
+
+ private void checkConfigs(NavigableMap<byte[], byte[]> configValuesResult,
+ Map<String, String> conf) throws IOException {
+
+ assertEquals(conf.size(), configValuesResult.size());
+ byte[] columnName;
+ for (String key : conf.keySet()) {
+ columnName = Bytes.toBytes(key);
+ assertTrue(configValuesResult.containsKey(columnName));
+ byte[] value = configValuesResult.get(columnName);
+ assertNotNull(value);
+ assertEquals(conf.get(key), GenericObjectMapper.read(value));
+ }
+ }
+
+ private void checkRelatedEntities(Map<String, Set<String>> isRelatedTo,
+ NavigableMap<byte[], byte[]> infoValues, byte[] columnPrefix)
+ throws IOException {
+
+ for (String key : isRelatedTo.keySet()) {
+ byte[] columnName = TimelineWriterUtils.join(
+ TimelineEntitySchemaConstants.ROW_KEY_SEPARATOR_BYTES, columnPrefix,
+ Bytes.toBytes(key));
+
+ byte[] value = infoValues.get(columnName);
+ assertNotNull(value);
+ String isRelatedToEntities = GenericObjectMapper.read(value).toString();
+ assertNotNull(isRelatedToEntities);
+ assertEquals(
+ TimelineWriterUtils.getValueAsString(
+ TimelineEntitySchemaConstants.ROW_KEY_SEPARATOR,
+ isRelatedTo.get(key)), isRelatedToEntities);
+ }
+ }
+
+ private boolean isRowKeyCorrect(byte[] rowKey, String cluster, String user,
+ String flow, Long runid, String appName, TimelineEntity te) {
+
+ byte[][] rowKeyComponents = TimelineWriterUtils.split(rowKey,
+ TimelineEntitySchemaConstants.ROW_KEY_SEPARATOR_BYTES);
+
+ assertTrue(rowKeyComponents.length == 7);
+ assertEquals(user, Bytes.toString(rowKeyComponents[0]));
+ assertEquals(cluster, Bytes.toString(rowKeyComponents[1]));
+ assertEquals(flow, Bytes.toString(rowKeyComponents[2]));
+ assertEquals(TimelineWriterUtils.encodeRunId(runid),
+ Bytes.toLong(rowKeyComponents[3]));
+ assertEquals(TimelineWriterUtils.cleanse(appName), Bytes.toString(rowKeyComponents[4]));
+ assertEquals(te.getType(), Bytes.toString(rowKeyComponents[5]));
+ assertEquals(te.getId(), Bytes.toString(rowKeyComponents[6]));
+ return true;
+ }
+
+ @AfterClass
+ public static void tearDownAfterClass() throws Exception {
+ util.shutdownMiniCluster();
+ }
+}