创建Segment过程概述
Peon进程由middle manager进程启动
提供http接口接收原始数据
将一行行的数据做merge，当行数达到maxRowsInMemory（当前集群配置是50000）或者intermediatePersistPeriod（10分钟），将内存中的数据序列化成Segment文件，持久化到本地磁盘
达到时间窗口后，停止接收数据，并将上步创建的segment合并成一个大的segment
新创建的segment通过hdfs移交到历史节点
启动Peon进程
middleManager接收到overlord分配过来的任务后，创建线程，设置jvm的命令并执行，包括classpath，堆内堆外内存设置，druid端口(用与peon进程对外的查询接口，这里分配的端口导致有端口被占用的bug)
任务json配置等准备工作。
执行jvm命令后，当前线程将一直等待直到任务完成。
Peon进程和导入相关的启动过程是：
- 创建接收原始数据的http接口，当前集群配置的EventReceiverRirehoseFactory
- 创建并执行,ExecutorLifeCycle根据json配置创建任务RealTimeIndexTask，创建lock文件对任务加锁，将任务交给ThreadPoolTaskRunner
@LifecycleStart
public void start() throws InterruptedException
{
final File taskFile = Preconditions.checkNotNull(taskExecutorConfig.getTaskFile(), "taskFile");
final File statusFile = Preconditions.checkNotNull(taskExecutorConfig.getStatusFile(), "statusFile");
final InputStream parentStream = Preconditions.checkNotNull(taskExecutorConfig.getParentStream(), "parentStream");
try {
task = jsonMapper.readValue(taskFile, Task.class);
log.info(
"Running with task: %s",
jsonMapper.writerWithDefaultPrettyPrinter().writeValueAsString(task)
);
}
catch (IOException e) {
throw Throwables.propagate(e);
}
// Avoid running the same task twice on the same machine by locking the task base directory.
final File taskLockFile = taskConfig.getTaskLockFile(task.getId());
try {
synchronized (this) {
if (taskLockChannel == null && taskLockFileLock == null) {
taskLockChannel = FileChannel.open(
taskLockFile.toPath(),
StandardOpenOption.CREATE,
StandardOpenOption.WRITE
);
log.info("Attempting to lock file[%s].", taskLockFile);
final long startLocking = System.currentTimeMillis();
final long timeout = DateTimes.utc(startLocking).plus(taskConfig.getDirectoryLockTimeout()).getMillis();
while (taskLockFileLock == null && System.currentTimeMillis() < timeout) {
taskLockFileLock = taskLockChannel.tryLock();
if (taskLockFileLock == null) {
Thread.sleep(100);
}
}
if (taskLockFileLock == null) {
throw new ISE("Could not acquire lock file[%s] within %,dms.", taskLockFile, timeout - startLocking);
} else {
log.info("Acquired lock file[%s] in %,dms.", taskLockFile, System.currentTimeMillis() - startLocking);
}
} else {
throw new ISE("Already started!");
}
}
}
catch (IOException e) {
throw Throwables.propagate(e);
}
if (taskExecutorConfig.isParentStreamDefined()) {
// Spawn monitor thread to keep a watch on parent's stdin
// If stdin reaches eof, the parent is gone, and we should shut down
parentMonitorExec.submit(
new Runnable()
{
@Override
public void run()
{
try {
while (parentStream.read() != -1) {
// Toss the byte
}
}
catch (Exception e) {
log.error(e, "Failed to read from stdin");
}
// Kind of gross, but best way to kill the JVM as far as I know
log.info("Triggering JVM shutdown.");
System.exit(2);
}
}
);
}
// Won't hurt in remote mode, and is required for setting up locks in local mode:
try {
if (!task.isReady(taskActionClientFactory.create(task))) {
throw new ISE("Task[%s] is not ready to run yet!", task.getId());
}
}
catch (Exception e) {
throw new ISE(e, "Failed to run task[%s] isReady", task.getId());
}
statusFuture = Futures.transform(
taskRunner.run(task),
new Function<TaskStatus, TaskStatus>()
{
@Override
public TaskStatus apply(TaskStatus taskStatus)
{
try {
log.info(
"Task completed with status: %s",
jsonMapper.writerWithDefaultPrettyPrinter().writeValueAsString(taskStatus)
);
final File statusFileParent = statusFile.getParentFile();
if (statusFileParent != null) {
statusFileParent.mkdirs();
}
jsonMapper.writeValue(statusFile, taskStatus);
return taskStatus;
}
catch (Exception e) {
throw Throwables.propagate(e);
}
}
}
);
}
ThreadPoolTaskRunner
runner.png
数据导入
Peon进程启动后，在RealtimeIndexTask.run()方法中，完成任务的执行。
- RealtimePlumber.startJob()完成任务的初始化配置，从磁盘加载已有的segment临时文件，启动持久化，merge，推送segment等线程
- EventReceiverFirehose接收数据行，OnheapIncrementalIndex创建索引
启动任务进程
导入数据到OnheapIncrementalIndex
EventReceiverFirehose提供http接口/push-events，接收tranquitiy提交的批量数据
EventReceiverFirehose 对数据处理的逻辑是：
对接收的数据进行解析，解析异常直接报错失败，对解析后的数据行接入到阻塞队列。
队列大小默认10万，如果任务线程队列消费数据不及时，接口会阻塞。
POST
@Path("/push-events")
@Consumes({MediaType.APPLICATION_JSON, SmileMediaTypes.APPLICATION_JACKSON_SMILE})
@Produces({MediaType.APPLICATION_JSON, SmileMediaTypes.APPLICATION_JACKSON_SMILE})
public Response addAll(
InputStream in,
@Context final HttpServletRequest req
)
{
ObjectMapper objectMapper = isSmile ? smileMapper : jsonMapper;
Optional<Response> producerSequenceResponse = checkProducerSequence(req, reqContentType, objectMapper);
if (producerSequenceResponse.isPresent()) {
return producerSequenceResponse.get();
}
CountingInputStream countingInputStream = new CountingInputStream(in);
Collection<Map<String, Object>> events = null;
try {
events = objectMapper.readValue(
countingInputStream, new TypeReference<Collection<Map<String, Object>>>()
{
}
);
}
catch (IOException e) {
return Response.serverError().entity(ImmutableMap.<String, Object>of("error", e.getMessage())).build();
}
finally {
bytesReceived.addAndGet(countingInputStream.getCount());
}
log.debug("Adding %,d events to firehose: %s", events.size(), serviceName);
final List<InputRow> rows = Lists.newArrayList();
for (final Map<String, Object> event : events) {
// Might throw an exception. We'd like that to happen now, instead of while adding to the row buffer.
rows.addAll(parser.parseBatch(event));
}
try {
addRows(rows);
return Response.ok(
objectMapper.writeValueAsString(ImmutableMap.of("eventCount", events.size())),
contentType
).build();
}
catch (InterruptedException e) {
...;
}
}
public void addRows(Iterable<InputRow> rows) throws InterruptedException
{
for (final InputRow row : rows) {
boolean added = false;
while (!closed && !added) {
added = buffer.offer(row, 500, TimeUnit.MILLISECONDS);
if (!added) {
long currTime = System.currentTimeMillis();
long lastTime = lastBufferAddFailMsgTime.get();
if (currTime - lastTime > 10000 && lastBufferAddFailMsgTime.compareAndSet(lastTime, currTime)) {
log.warn("Failed to add event to buffer with current size [%s] . Retrying...", buffer.size());
}
}
}
if (!added) {
throw new IllegalStateException("Cannot add events to closed firehose!");
}
}
}
RealtimeIndexTask持有上部的firehose的实例，消费缓存对垒，交给realtimePlumber处理：“
1 根据数据行的时间戳获取sink，每个interval对应一个sink，一般一个peon进程就一个sink；
sink中有多个Firehydrant，有一个负责响应查询到增量导入数据，其余只是负责查询
2.添加数据行到sink的房前firehydrant，由Firehydrant的onheapIndecrementalindex完成增量索引创建
3.判读当前的firehydrant中已有的数据行数，如果达到配置maxRowsinmemory，或者处理时间超过配置的intermediatePersistperiod，把当前firehydrant数据持久化到磁盘
public int add(InputRow row, Supplier<Committer> committerSupplier) throws IndexSizeExceededException
{
long messageTimestamp = row.getTimestampFromEpoch();
final Sink sink = getSink(messageTimestamp);
metrics.reportMessageMaxTimestamp(messageTimestamp);
if (sink == null) {
return -1;
}
final IncrementalIndexAddResult addResult = sink.add(row, false);
if (config.isReportParseExceptions() && addResult.getParseException() != null) {
throw addResult.getParseException();
}
if (!sink.canAppendRow() || System.currentTimeMillis() > nextFlush) {
persist(committerSupplier.get());
}
return addResult.getRowCount();
}
一个sink包含了最终生成的segment数据，一个segment数据比较大，不合适都放在内存中，在创建segment中，会达到一定数据后持久化到磁盘，会有小小的segment的问题件，有一个个的Firehydrant表示，已经持久化到磁盘的数据对应的Firehydrant对象只负责响应查询，Sink中还有当前Firehydrant，既响应查询也不断将数据加入到Firehydrant对象中。
如果当前数据行时间字所在的interval没有对应的sink，就会创建新的sink对象。创建Sink的过程中，就会创建当前的FireHydrant对象
private FireHydrant makeNewCurrIndex(long minTimestamp, DataSchema schema)
{
final IncrementalIndexSchema indexSchema = new IncrementalIndexSchema.Builder()
.withMinTimestamp(minTimestamp)
.withTimestampSpec(schema.getParser())
.withQueryGranularity(schema.getGranularitySpec().getQueryGranularity())
.withDimensionsSpec(schema.getParser())
.withMetrics(schema.getAggregators())
.withRollup(schema.getGranularitySpec().isRollup())
.build();
//创建OnheapIncrementalIndex对象
final IncrementalIndex newIndex = new IncrementalIndex.Builder()
.setIndexSchema(indexSchema)
.setReportParseExceptions(reportParseExceptions)
.setMaxRowCount(maxRowsInMemory)
.buildOnheap();
final FireHydrant old;
synchronized (hydrantLock) {
if (writable) {
old = currHydrant;
int newCount = 0;
int numHydrants = hydrants.size();
if (numHydrants > 0) {
FireHydrant lastHydrant = hydrants.get(numHydrants - 1);
newCount = lastHydrant.getCount() + 1;
if (!indexSchema.getDimensionsSpec().hasCustomDimensions()) {
Map<String, ColumnCapabilitiesImpl> oldCapabilities;
if (lastHydrant.hasSwapped()) {
oldCapabilities = Maps.newHashMap();
ReferenceCountingSegment segment = lastHydrant.getIncrementedSegment();
try {
QueryableIndex oldIndex = segment.asQueryableIndex();
for (String dim : oldIndex.getAvailableDimensions()) {
dimOrder.add(dim);
oldCapabilities.put(dim, (ColumnCapabilitiesImpl) oldIndex.getColumn(dim).getCapabilities());
}
}
finally {
segment.decrement();
}
} else {
IncrementalIndex oldIndex = lastHydrant.getIndex();
dimOrder.addAll(oldIndex.getDimensionOrder());
oldCapabilities = oldIndex.getColumnCapabilities();
}
newIndex.loadDimensionIterable(dimOrder, oldCapabilities);
}
}
currHydrant = new FireHydrant(newIndex, newCount, getSegment().getIdentifier());
if (old != null) {
numRowsExcludingCurrIndex.addAndGet(old.getIndex().size());
}
//新创建的FireHydrant加入到Sink
hydrants.add(currHydrant);
} else {
// Oops, someone called finishWriting while we were making this new index.
newIndex.close();
throw new ISE("finishWriting() called during swap");
}
}
return old;
}
数据行的添加最终是在OnheapIncrementalIndex对象的addToFacts方法完成：
OnheapIncrementalIndex实现可以理解成有一个Map对象以维度列和时间列TimeAndDims作为key，指标列作为value，当新的数据行加入时，通过key（TimeAndDims）确认对应的Aggregator聚合器对象，
维度值出现null值，作为一个单独的值加到维度字典编码中
Aggregator聚合器对象完成对数据行的累加操作
聚合时，使用ColumnSelectorFactory获取每行的指标值，和查询时通过游标获取列值不同，这里通过threadlocal方式获取，每次聚合前IncrementalIndex将数据InputRow放入到threadlocal
猜测这样实现而不是聚合时直接传入指标值的方式，主要是因为指标值聚合前需要类型转换和值的各种转换，这块逻辑主要在ColumnSelector完成，
这样，IncrementalIndex只负责将数据添加到threadlocal，Aggregator只需要从ColumnSelector获取要聚合的数据RollupFactsHolder（跳表）记录了维度值->行号，aggregators记录了行号->指标值数组的映射
protected AddToFactsResult addToFacts(
AggregatorFactory[] metrics,
boolean deserializeComplexMetrics,
boolean reportParseExceptions,
InputRow row,
AtomicInteger numEntries,
TimeAndDims key,
ThreadLocal<InputRow> rowContainer,
Supplier<InputRow> rowSupplier,
boolean skipMaxRowsInMemoryCheck
) throws IndexSizeExceededException
{
List<String> parseExceptionMessages;
final int priorIndex = facts.getPriorIndex(key);
Aggregator[] aggs;
if (TimeAndDims.EMPTY_ROW_INDEX != priorIndex) {
aggs = concurrentGet(priorIndex);
parseExceptionMessages = doAggregate(metrics, aggs, rowContainer, row);
} else {
aggs = new Aggregator[metrics.length];
factorizeAggs(metrics, aggs, rowContainer, row);
parseExceptionMessages = doAggregate(metrics, aggs, rowContainer, row);
final int rowIndex = indexIncrement.getAndIncrement();
concurrentSet(rowIndex, aggs);
// Last ditch sanity checks
if (numEntries.get() >= maxRowCount
&& facts.getPriorIndex(key) == TimeAndDims.EMPTY_ROW_INDEX
&& !skipMaxRowsInMemoryCheck) {
throw new IndexSizeExceededException("Maximum number of rows [%d] reached", maxRowCount);
}
final int prev = facts.putIfAbsent(key, rowIndex);
if (TimeAndDims.EMPTY_ROW_INDEX == prev) {
numEntries.incrementAndGet();
} else {
// We lost a race
aggs = concurrentGet(prev);
parseExceptionMessages = doAggregate(metrics, aggs, rowContainer, row);
// Free up the misfire
concurrentRemove(rowIndex);
// This is expected to occur ~80% of the time in the worst scenarios
}
}
return new AddToFactsResult(numEntries.get(), parseExceptionMessages);
}
前FireHydrant中已有的数据行数如果达到配置的maxRowsInMemory，或者处理时间超过配置的intermediatePersistPeriod，将把当前FireHydrant数据持久化到磁盘
任务移交
任务移交.png
网友评论