flume-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From hshreedha...@apache.org
Subject flume git commit: FLUME-2875. Allow RollingFileSink to specify a file prefix and a file extension.
Date Tue, 09 Feb 2016 01:42:23 GMT
Repository: flume
Updated Branches:
  refs/heads/flume-1.7 a6b55f183 -> b03ad7168


FLUME-2875. Allow RollingFileSink to specify a file prefix and a file extension.

(Ralph Goers via Hari)


Project: http://git-wip-us.apache.org/repos/asf/flume/repo
Commit: http://git-wip-us.apache.org/repos/asf/flume/commit/b03ad716
Tree: http://git-wip-us.apache.org/repos/asf/flume/tree/b03ad716
Diff: http://git-wip-us.apache.org/repos/asf/flume/diff/b03ad716

Branch: refs/heads/flume-1.7
Commit: b03ad71685b1032083145ee0a48ac5e4de8570a3
Parents: a6b55f1
Author: Hari Shreedharan <hshreedharan@apache.org>
Authored: Mon Feb 8 17:40:35 2016 -0800
Committer: Hari Shreedharan <hshreedharan@apache.org>
Committed: Mon Feb 8 17:41:44 2016 -0800

----------------------------------------------------------------------
 .../formatter/output/DefaultPathManager.java    | 108 +++++++++++++++++++
 .../flume/formatter/output/PathManager.java     |  63 ++++-------
 .../formatter/output/PathManagerFactory.java    |  82 ++++++++++++++
 .../flume/formatter/output/PathManagerType.java |  43 ++++++++
 .../formatter/output/RollTimePathManager.java   |  66 ++++++++++++
 .../org/apache/flume/sink/RollingFileSink.java  |   8 +-
 .../apache/flume/sink/TestRollingFileSink.java  | 106 ++++++++++++++++++
 flume-ng-doc/sphinx/FlumeUserGuide.rst          |  53 ++++-----
 8 files changed, 460 insertions(+), 69 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flume/blob/b03ad716/flume-ng-core/src/main/java/org/apache/flume/formatter/output/DefaultPathManager.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/formatter/output/DefaultPathManager.java
b/flume-ng-core/src/main/java/org/apache/flume/formatter/output/DefaultPathManager.java
new file mode 100644
index 0000000..176db7f
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/formatter/output/DefaultPathManager.java
@@ -0,0 +1,108 @@
+/*
+ * 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.flume.formatter.output;
+
+import java.io.File;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.flume.Context;
+
+public class DefaultPathManager implements PathManager {
+
+  private long seriesTimestamp;
+  private File baseDirectory;
+  private AtomicInteger fileIndex;
+  private String filePrefix;
+  private String extension;
+
+  private static final String DEFAULT_FILE_PREFIX = "";
+  private static final String DEFAULT_FILE_EXTENSION = "";
+  private static final String FILE_EXTENSION = "extension";
+  private static final String FILE_PREFIX = "prefix";
+
+  protected File currentFile;
+
+  public DefaultPathManager(Context context) {
+    filePrefix = context.getString(FILE_PREFIX, DEFAULT_FILE_PREFIX);
+    extension = context.getString(FILE_EXTENSION, DEFAULT_FILE_EXTENSION);
+    seriesTimestamp = System.currentTimeMillis();
+    fileIndex = new AtomicInteger();
+  }
+
+  @Override
+  public File nextFile() {
+    StringBuilder sb = new StringBuilder();
+    sb.append(filePrefix).append(seriesTimestamp).append("-");
+    sb.append(fileIndex.incrementAndGet());
+    if (extension.length() > 0) {
+        sb.append(".").append(extension);
+    }
+    currentFile = new File(baseDirectory, sb.toString());
+
+    return currentFile;
+  }
+
+  @Override
+  public File getCurrentFile() {
+    if (currentFile == null) {
+      return nextFile();
+    }
+
+    return currentFile;
+  }
+
+  @Override
+  public void rotate() {
+    currentFile = null;
+  }
+
+  @Override
+  public File getBaseDirectory() {
+    return baseDirectory;
+  }
+
+  @Override
+  public void setBaseDirectory(File baseDirectory) {
+    this.baseDirectory = baseDirectory;
+  }
+
+  public long getSeriesTimestamp() {
+    return seriesTimestamp;
+  }
+
+  public String getPrefix() {
+    return filePrefix;
+  }
+
+  public String getExtension() {
+    return extension;
+  }
+
+  public AtomicInteger getFileIndex() {
+    return fileIndex;
+  }
+
+  public static class Builder implements PathManager.Builder {
+    @Override
+    public PathManager build(Context context) {
+      return new DefaultPathManager(context);
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/flume/blob/b03ad716/flume-ng-core/src/main/java/org/apache/flume/formatter/output/PathManager.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/formatter/output/PathManager.java
b/flume-ng-core/src/main/java/org/apache/flume/formatter/output/PathManager.java
index 933cc94..5a3066a 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/formatter/output/PathManager.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/formatter/output/PathManager.java
@@ -16,58 +16,35 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-
 package org.apache.flume.formatter.output;
 
 import java.io.File;
-import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.flume.Context;
 
-public class PathManager {
+/**
+ * Creates the files used by the RollingFileSink.
+ */
+public interface PathManager {
+    /**
+     * {@link Context} prefix
+     */
+    public static String CTX_PREFIX = "pathManager.";
 
-  private long seriesTimestamp;
-  private File baseDirectory;
-  private AtomicInteger fileIndex;
+    File nextFile();
 
-  private File currentFile;
+    File getCurrentFile();
 
-  public PathManager() {
-    seriesTimestamp = System.currentTimeMillis();
-    fileIndex = new AtomicInteger();
-  }
+    void rotate();
 
-  public File nextFile() {
-    currentFile = new File(baseDirectory, seriesTimestamp + "-"
-        + fileIndex.incrementAndGet());
+    File getBaseDirectory();
 
-    return currentFile;
-  }
+    void setBaseDirectory(File baseDirectory);
 
-  public File getCurrentFile() {
-    if (currentFile == null) {
-      return nextFile();
+    /**
+     * Knows how to construct this path manager.<br/>
+     * <b>Note: Implementations MUST provide a public a no-arg constructor.</b>
+     */
+    public interface Builder {
+        public PathManager build(Context context);
     }
-
-    return currentFile;
-  }
-
-  public void rotate() {
-    currentFile = null;
-  }
-
-  public File getBaseDirectory() {
-    return baseDirectory;
-  }
-
-  public void setBaseDirectory(File baseDirectory) {
-    this.baseDirectory = baseDirectory;
-  }
-
-  public long getSeriesTimestamp() {
-    return seriesTimestamp;
-  }
-
-  public AtomicInteger getFileIndex() {
-    return fileIndex;
-  }
-
 }

http://git-wip-us.apache.org/repos/asf/flume/blob/b03ad716/flume-ng-core/src/main/java/org/apache/flume/formatter/output/PathManagerFactory.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/formatter/output/PathManagerFactory.java
b/flume-ng-core/src/main/java/org/apache/flume/formatter/output/PathManagerFactory.java
new file mode 100644
index 0000000..4dbe083
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/formatter/output/PathManagerFactory.java
@@ -0,0 +1,82 @@
+/*
+ * 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.flume.formatter.output;
+
+import com.google.common.base.Preconditions;
+import java.util.Locale;
+import org.apache.flume.Context;
+import org.apache.flume.FlumeException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Create PathManager instances.
+ */
+public class PathManagerFactory {
+    private static final Logger logger = LoggerFactory.getLogger(PathManagerFactory.class);
+
+    public static PathManager getInstance(String managerType, Context context) {
+
+        Preconditions.checkNotNull(managerType, "path manager type must not be null");
+
+        // try to find builder class in enum of known output serializers
+        PathManagerType type;
+        try {
+            type = PathManagerType.valueOf(managerType.toUpperCase(Locale.ENGLISH));
+        } catch (IllegalArgumentException e) {
+            logger.debug("Not in enum, loading builder class: {}", managerType);
+            type = PathManagerType.OTHER;
+        }
+        Class<? extends PathManager.Builder> builderClass = type.getBuilderClass();
+
+        // handle the case where they have specified their own builder in the config
+        if (builderClass == null) {
+            try {
+                Class c = Class.forName(managerType);
+                if (c != null && PathManager.Builder.class.isAssignableFrom(c)) {
+                    builderClass = (Class<? extends PathManager.Builder>) c;
+                } else {
+                    String errMessage = "Unable to instantiate Builder from " +
+                            managerType + ": does not appear to implement " +
+                            PathManager.Builder.class.getName();
+                    throw new FlumeException(errMessage);
+                }
+            } catch (ClassNotFoundException ex) {
+                logger.error("Class not found: " + managerType, ex);
+                throw new FlumeException(ex);
+            }
+        }
+
+        // build the builder
+        PathManager.Builder builder;
+        try {
+            builder = builderClass.newInstance();
+        } catch (InstantiationException ex) {
+            String errMessage = "Cannot instantiate builder: " + managerType;
+            logger.error(errMessage, ex);
+            throw new FlumeException(errMessage, ex);
+        } catch (IllegalAccessException ex) {
+            String errMessage = "Cannot instantiate builder: " + managerType;
+            logger.error(errMessage, ex);
+            throw new FlumeException(errMessage, ex);
+        }
+
+        return builder.build(context);
+    }
+}

http://git-wip-us.apache.org/repos/asf/flume/blob/b03ad716/flume-ng-core/src/main/java/org/apache/flume/formatter/output/PathManagerType.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/formatter/output/PathManagerType.java
b/flume-ng-core/src/main/java/org/apache/flume/formatter/output/PathManagerType.java
new file mode 100644
index 0000000..4f1fa93
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/formatter/output/PathManagerType.java
@@ -0,0 +1,43 @@
+/*
+ * 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.flume.formatter.output;
+
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+
+/**
+ *
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public enum PathManagerType {
+    DEFAULT(DefaultPathManager.Builder.class),
+    ROLLTIME(RollTimePathManager.Builder.class),
+    OTHER(null);
+
+    private final Class<? extends PathManager.Builder> builderClass;
+
+    PathManagerType(Class<? extends PathManager.Builder> builderClass) {
+        this.builderClass = builderClass;
+    }
+
+    public Class<? extends PathManager.Builder> getBuilderClass() {
+        return builderClass;
+    }
+}

http://git-wip-us.apache.org/repos/asf/flume/blob/b03ad716/flume-ng-core/src/main/java/org/apache/flume/formatter/output/RollTimePathManager.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/formatter/output/RollTimePathManager.java
b/flume-ng-core/src/main/java/org/apache/flume/formatter/output/RollTimePathManager.java
new file mode 100644
index 0000000..6883a9c
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/formatter/output/RollTimePathManager.java
@@ -0,0 +1,66 @@
+/*
+ * 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.flume.formatter.output;
+
+import java.io.File;
+
+import org.apache.flume.Context;
+import org.joda.time.LocalDateTime;
+import org.joda.time.format.DateTimeFormat;
+import org.joda.time.format.DateTimeFormatter;
+
+/**
+ *
+ */
+public class RollTimePathManager extends DefaultPathManager {
+
+    private final DateTimeFormatter formatter = DateTimeFormat.forPattern("yyyyMMddHHmmss");
+    private String lastRoll;
+
+    public RollTimePathManager(Context context) {
+        super(context);
+    }
+
+    @Override
+    public File nextFile() {
+        StringBuilder sb = new StringBuilder();
+        String date = formatter.print(LocalDateTime.now());
+        if (!date.equals(lastRoll)) {
+            getFileIndex().set(0);
+            lastRoll = date;
+        }
+        sb.append(getPrefix()).append(date).append("-");
+        sb.append(getFileIndex().incrementAndGet());
+        if (getExtension().length() > 0) {
+            sb.append(".").append(getExtension());
+        }
+        currentFile = new File(getBaseDirectory(), sb.toString());
+
+        return currentFile;
+    }
+
+    public static class Builder implements PathManager.Builder {
+        @Override
+        public PathManager build(Context context) {
+            return new RollTimePathManager(context);
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/flume/blob/b03ad716/flume-ng-core/src/main/java/org/apache/flume/sink/RollingFileSink.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/sink/RollingFileSink.java b/flume-ng-core/src/main/java/org/apache/flume/sink/RollingFileSink.java
index 9cb3370..b97d404 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/sink/RollingFileSink.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/sink/RollingFileSink.java
@@ -33,6 +33,7 @@ import org.apache.flume.EventDeliveryException;
 import org.apache.flume.Transaction;
 import org.apache.flume.conf.Configurable;
 import org.apache.flume.formatter.output.PathManager;
+import org.apache.flume.formatter.output.PathManagerFactory;
 import org.apache.flume.instrumentation.SinkCounter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -66,13 +67,13 @@ public class RollingFileSink extends AbstractSink implements Configurable
{
   private volatile boolean shouldRotate;
 
   public RollingFileSink() {
-    pathController = new PathManager();
     shouldRotate = false;
   }
 
   @Override
   public void configure(Context context) {
 
+    String pathManagerType = context.getString("sink.pathManager", "DEFAULT");
     String directory = context.getString("sink.directory");
     String rollInterval = context.getString("sink.rollInterval");
 
@@ -81,6 +82,11 @@ public class RollingFileSink extends AbstractSink implements Configurable
{
         new Context(context.getSubProperties("sink." +
             EventSerializer.CTX_PREFIX));
 
+    Context pathManagerContext =
+              new Context(context.getSubProperties("sink." +
+                      PathManager.CTX_PREFIX));
+    pathController = PathManagerFactory.getInstance(pathManagerType, pathManagerContext);
+
     Preconditions.checkArgument(directory != null, "Directory may not be null");
     Preconditions.checkNotNull(serializerType, "Serializer type is undefined");
 

http://git-wip-us.apache.org/repos/asf/flume/blob/b03ad716/flume-ng-core/src/test/java/org/apache/flume/sink/TestRollingFileSink.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/sink/TestRollingFileSink.java b/flume-ng-core/src/test/java/org/apache/flume/sink/TestRollingFileSink.java
index 07fa644..bf4ed1f 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/sink/TestRollingFileSink.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/sink/TestRollingFileSink.java
@@ -172,4 +172,110 @@ public class TestRollingFileSink {
       reader.close();
     }
   }
+
+  @Test
+  public void testAppend3() throws InterruptedException, LifecycleException, EventDeliveryException,
IOException {
+    File tmpDir = new File("target/tmpLog");
+    tmpDir.mkdirs();
+    cleanDirectory(tmpDir);
+    Context context = new Context();
+
+    context.put("sink.directory", "target/tmpLog");
+    context.put("sink.rollInterval", "0");
+    context.put("sink.batchSize", "1");
+    context.put("sink.pathManager.prefix", "test3-");
+    context.put("sink.pathManager.extension", "txt");
+
+    Configurables.configure(sink, context);
+
+    Channel channel = new PseudoTxnMemoryChannel();
+    Configurables.configure(channel, context);
+
+    sink.setChannel(channel);
+    sink.start();
+
+    for (int i = 0; i < 10; i++) {
+      Event event = new SimpleEvent();
+
+      event.setBody(("Test event " + i).getBytes());
+
+      channel.put(event);
+      sink.process();
+
+      Thread.sleep(500);
+    }
+
+    sink.stop();
+
+    for (String file : sink.getDirectory().list()) {
+      BufferedReader reader = new BufferedReader(new FileReader(new File(sink.getDirectory(),
file)));
+
+      String lastLine = null;
+      String currentLine = null;
+
+      while ((currentLine = reader.readLine()) != null) {
+        lastLine = currentLine;
+        logger.debug("Produced file:{} lastLine:{}", file, lastLine);
+      }
+
+      reader.close();
+    }
+  }
+
+  @Test
+  public void testRollTime() throws InterruptedException, LifecycleException, EventDeliveryException,
IOException {
+    File tmpDir = new File("target/tempLog");
+    tmpDir.mkdirs();
+    cleanDirectory(tmpDir);
+    Context context = new Context();
+
+    context.put("sink.directory", "target/tempLog/");
+    context.put("sink.rollInterval", "1");
+    context.put("sink.batchSize", "1");
+    context.put("sink.pathManager", "rolltime");
+    context.put("sink.pathManager.prefix", "test4-");
+    context.put("sink.pathManager.extension", "txt");
+
+    Configurables.configure(sink, context);
+
+    Channel channel = new PseudoTxnMemoryChannel();
+    Configurables.configure(channel, context);
+
+    sink.setChannel(channel);
+    sink.start();
+
+    for (int i = 0; i < 10; i++) {
+      Event event = new SimpleEvent();
+
+      event.setBody(("Test event " + i).getBytes());
+
+      channel.put(event);
+      sink.process();
+
+      Thread.sleep(500);
+    }
+
+    sink.stop();
+
+    for (String file : sink.getDirectory().list()) {
+      BufferedReader reader = new BufferedReader(new FileReader(new File(sink.getDirectory(),
file)));
+
+      String lastLine = null;
+      String currentLine = null;
+
+      while ((currentLine = reader.readLine()) != null) {
+        lastLine = currentLine;
+        logger.debug("Produced file:{} lastLine:{}", file, lastLine);
+      }
+
+      reader.close();
+    }
+  }
+
+  private void cleanDirectory(File dir) {
+    File[] files = dir.listFiles();
+    for (File file : files) {
+      file.delete();
+    }
+  }
 }

http://git-wip-us.apache.org/repos/asf/flume/blob/b03ad716/flume-ng-doc/sphinx/FlumeUserGuide.rst
----------------------------------------------------------------------
diff --git a/flume-ng-doc/sphinx/FlumeUserGuide.rst b/flume-ng-doc/sphinx/FlumeUserGuide.rst
index 0f8461d..423e0cf 100644
--- a/flume-ng-doc/sphinx/FlumeUserGuide.rst
+++ b/flume-ng-doc/sphinx/FlumeUserGuide.rst
@@ -28,22 +28,22 @@ Apache Flume is a distributed, reliable, and available system for efficiently
 collecting, aggregating and moving large amounts of log data from many
 different sources to a centralized data store.
 
-The use of Apache Flume is not only restricted to log data aggregation. 
+The use of Apache Flume is not only restricted to log data aggregation.
 Since data sources are customizable, Flume can be used to transport massive quantities
-of event data including but not limited to network traffic data, social-media-generated data,

+of event data including but not limited to network traffic data, social-media-generated data,
 email messages and pretty much any data source possible.
 
 Apache Flume is a top level project at the Apache Software Foundation.
 
 There are currently two release code lines available, versions 0.9.x and 1.x.
 
-Documentation for the 0.9.x track is available at 
+Documentation for the 0.9.x track is available at
 `the Flume 0.9.x User Guide <http://archive.cloudera.com/cdh/3/flume/UserGuide/>`_.
 
 This documentation applies to the 1.4.x track.
 
-New and existing users are encouraged to use the 1.x releases so as to 
-leverage the performance improvements and configuration flexibilities available 
+New and existing users are encouraged to use the 1.x releases so as to
+leverage the performance improvements and configuration flexibilities available
 in the latest architecture.
 
 
@@ -1153,7 +1153,7 @@ Twitter 1% firehose Source (experimental)
 
 Experimental source that connects via Streaming API to the 1% sample twitter
 firehose, continously downloads tweets, converts them to Avro format and
-sends Avro events to a downstream Flume sink. Requires the consumer and 
+sends Avro events to a downstream Flume sink. Requires the consumer and
 access tokens and secrets of a Twitter developer account.
 Required properties are in **bold**.
 
@@ -1165,7 +1165,7 @@ Property Name          Default      Description
 **consumerKey**        --           OAuth consumer key
 **consumerSecret**     --           OAuth consumer secret
 **accessToken**        --           OAuth access token
-**accessTokenSecret**  --           OAuth toekn secret 
+**accessTokenSecret**  --           OAuth toekn secret
 maxBatchSize           1000         Maximum number of twitter messages to put in a single
batch
 maxBatchDurationMillis 1000         Maximum number of milliseconds to wait before closing
a batch
 ====================== ===========  ===================================================
@@ -2119,16 +2119,19 @@ File Roll Sink
 Stores events on the local filesystem.
 Required properties are in **bold**.
 
-===================  =======  ======================================================================================================================
-Property Name        Default  Description
-===================  =======  ======================================================================================================================
-**channel**          --
-**type**             --       The component type name, needs to be ``file_roll``.
-**sink.directory**   --       The directory where files will be stored
-sink.rollInterval    30       Roll the file every 30 seconds. Specifying 0 will disable rolling
and cause all events to be written to a single file.
-sink.serializer      TEXT     Other possible options include ``avro_event`` or the FQCN of
an implementation of EventSerializer.Builder interface.
-batchSize            100
-===================  =======  ======================================================================================================================
+==========================  =======  ======================================================================================================================
+Property Name               Default  Description
+==========================  =======  ======================================================================================================================
+**channel**                 --
+**type**                    --       The component type name, needs to be ``file_roll``.
+**sink.directory**          --       The directory where files will be stored
+sink.pathManager            DEFAULT  The PathManager implementation to use.
+sink.pathManager.extension  --       The file extension if the default PathManager is used.
+sink.pathManager.prefix     --       A character string to add to the beginning of the file
name if the default PathManager is used
+sink.rollInterval           30       Roll the file every 30 seconds. Specifying 0 will disable
rolling and cause all events to be written to a single file.
+sink.serializer             TEXT     Other possible options include ``avro_event`` or the
FQCN of an implementation of EventSerializer.Builder interface.
+batchSize                   100
+==========================  =======  ======================================================================================================================
 
 Example for agent named a1:
 
@@ -2284,19 +2287,19 @@ This sink extracts data from Flume events, transforms it, and loads
it in near-r
 
 This sink is well suited for use cases that stream raw data into HDFS (via the HdfsSink)
and simultaneously extract, transform and load the same data into Solr (via MorphlineSolrSink).
In particular, this sink can process arbitrary heterogeneous raw data from disparate data
sources and turn it into a data model that is useful to Search applications.
 
-The ETL functionality is customizable using a `morphline configuration file <http://cloudera.github.io/cdk/docs/current/cdk-morphlines/index.html>`_
that defines a chain of transformation commands that pipe event records from one command to
another. 
+The ETL functionality is customizable using a `morphline configuration file <http://cloudera.github.io/cdk/docs/current/cdk-morphlines/index.html>`_
that defines a chain of transformation commands that pipe event records from one command to
another.
 
 Morphlines can be seen as an evolution of Unix pipelines where the data model is generalized
to work with streams of generic records, including arbitrary binary payloads. A morphline
command is a bit like a Flume Interceptor. Morphlines can be embedded into Hadoop components
such as Flume.
 
 Commands to parse and transform a set of standard data formats such as log files, Avro, CSV,
Text, HTML, XML, PDF, Word, Excel, etc. are provided out of the box, and additional custom
commands and parsers for additional data formats can be added as morphline plugins. Any kind
of data format can be indexed and any Solr documents for any kind of Solr schema can be generated,
and any custom ETL logic can be registered and executed.
 
-Morphlines manipulate continuous streams of records. The data model can be described as follows:
A record is a set of named fields where each field has an ordered list of one or more values.
A value can be any Java Object. That is, a record is essentially a hash table where each hash
table entry contains a String key and a list of Java Objects as values. (The implementation
uses Guava's ``ArrayListMultimap``, which is a ``ListMultimap``). Note that a field can have
multiple values and any two records need not use common field names. 
+Morphlines manipulate continuous streams of records. The data model can be described as follows:
A record is a set of named fields where each field has an ordered list of one or more values.
A value can be any Java Object. That is, a record is essentially a hash table where each hash
table entry contains a String key and a list of Java Objects as values. (The implementation
uses Guava's ``ArrayListMultimap``, which is a ``ListMultimap``). Note that a field can have
multiple values and any two records need not use common field names.
 
 This sink fills the body of the Flume event into the ``_attachment_body`` field of the morphline
record, as well as copies the headers of the Flume event into record fields of the same name.
The commands can then act on this data.
 
 Routing to a SolrCloud cluster is supported to improve scalability. Indexing load can be
spread across a large number of MorphlineSolrSinks for improved scalability. Indexing load
can be replicated across multiple MorphlineSolrSinks for high availability, for example using
Flume features such as Load balancing Sink Processor. MorphlineInterceptor can also help to
implement dynamic routing to multiple Solr collections (e.g. for multi-tenancy).
 
-The morphline and solr jars required for your environment must be placed in the lib directory
of the Apache Flume installation. 
+The morphline and solr jars required for your environment must be placed in the lib directory
of the Apache Flume installation.
 
 The type is the FQCN: org.apache.flume.sink.solr.morphline.MorphlineSolrSink
 
@@ -2334,11 +2337,11 @@ ElasticSearchSink
 ~~~~~~~~~~~~~~~~~
 
 This sink writes data to an elasticsearch cluster. By default, events will be written so
that the `Kibana <http://kibana.org>`_ graphical interface
-can display them - just as if `logstash <https://logstash.net>`_ wrote them. 
+can display them - just as if `logstash <https://logstash.net>`_ wrote them.
 
-The elasticsearch and lucene-core jars required for your environment must be placed in the
lib directory of the Apache Flume installation. 
+The elasticsearch and lucene-core jars required for your environment must be placed in the
lib directory of the Apache Flume installation.
 Elasticsearch requires that the major version of the client JAR match that of the server
and that both are running the same minor version
-of the JVM. SerializationExceptions will appear if this is incorrect. To 
+of the JVM. SerializationExceptions will appear if this is incorrect. To
 select the required version first determine the version of elasticsearch and the JVM version
the target cluster is running. Then select an elasticsearch client
 library which matches the major version. A 0.19.x client can talk to a 0.19.x cluster; 0.20.x
can talk to 0.20.x and 0.90.x can talk to 0.90.x. Once the
 elasticsearch version has been determined then read the pom.xml file to determine the correct
lucene-core JAR version to use. The Flume agent
@@ -2588,7 +2591,7 @@ Example for agent named a1:
   a1.channels.c1.transactionCapacity = 10000
   a1.channels.c1.byteCapacityBufferPercentage = 20
   a1.channels.c1.byteCapacity = 800000
-  
+
 
 JDBC Channel
 ~~~~~~~~~~~~
@@ -2796,7 +2799,7 @@ The disk store is managed using an embedded File channel. When the in-memory
que
 the file channel. This channel is ideal for flows that need high throughput of memory channel
during normal operation, but at the
 same time need the larger capacity of the file channel for better tolerance of intermittent
sink side outages or drop in drain rates.
 The throughput will reduce approximately to file channel speeds during such abnormal situations.
In case of an agent crash or restart,
-only the events stored on disk are recovered when the agent comes online. **This channel
is currently experimental and 
+only the events stored on disk are recovered when the agent comes online. **This channel
is currently experimental and
 not recommended for use in production.**
 
 Required properties are in **bold**. Please refer to file channel for additional required
properties.


Mime
View raw message