Apache Apex 是一个用于大数据流和批处理的统一平台 使用案例包括摄取、ETL、实时分析、警报和实时操作

/** * 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 com.datatorrent.stram.cli; import java.io.BufferedReader; import java.io.File; import java.io.FileInputStream; import java.io.FileNotFoundException; import java.io.FileReader; import java.io.IOException; import java.io.OutputStream; import java.io.PrintStream; import java.io.PrintWriter; import java.lang.management.ManagementFactory; import java.net.URI; import java.net.URISyntaxException; import java.net.URLEncoder; import java.security.PrivilegedExceptionAction; import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.Comparator; import java.util.Enumeration; import java.util.HashMap; import java.util.Iterator; import java.util.LinkedHashMap; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.Set; import java.util.TreeMap; import java.util.TreeSet; import javax.validation.constraints.NotNull; import javax.ws.rs.core.MediaType; import org.codehaus.jackson.map.ObjectMapper; import org.codehaus.jettison.json.JSONArray; import org.codehaus.jettison.json.JSONException; import org.codehaus.jettison.json.JSONObject; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.commons.cli.BasicParser; 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.OptionBuilder; import org.apache.commons.cli.Options; import org.apache.commons.cli.ParseException; import org.apache.commons.cli.PosixParser; import org.apache.commons.io.FileUtils; import org.apache.commons.io.IOUtils; import org.apache.commons.lang.ArrayUtils; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.apache.commons.lang.math.NumberUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.CommonConfigurationKeysPublic; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.security.UserGroupInformation; import org.apache.hadoop.yarn.api.records.ApplicationId; import org.apache.hadoop.yarn.api.records.ApplicationReport; import org.apache.hadoop.yarn.api.records.FinalApplicationStatus; import org.apache.hadoop.yarn.api.records.YarnApplicationState; import org.apache.hadoop.yarn.client.api.YarnClient; import org.apache.hadoop.yarn.conf.YarnConfiguration; import org.apache.hadoop.yarn.exceptions.YarnException; import org.apache.log4j.Appender; import org.apache.log4j.ConsoleAppender; import org.apache.log4j.Level; import org.apache.tools.ant.DirectoryScanner; import com.google.common.base.Preconditions; import com.sun.jersey.api.client.WebResource; import com.datatorrent.api.DAG.GenericOperator; import com.datatorrent.api.Operator; import com.datatorrent.api.StreamingApplication; import com.datatorrent.stram.StramUtils; import com.datatorrent.stram.client.AppPackage; import com.datatorrent.stram.client.AppPackage.AppInfo; import com.datatorrent.stram.client.AppPackage.PropertyInfo; import com.datatorrent.stram.client.ConfigPackage; import com.datatorrent.stram.client.DTConfiguration; import com.datatorrent.stram.client.DTConfiguration.Scope; import com.datatorrent.stram.client.RecordingsAgent; import com.datatorrent.stram.client.RecordingsAgent.RecordingInfo; import com.datatorrent.stram.client.StramAgent; import com.datatorrent.stram.client.StramAppLauncher; import com.datatorrent.stram.client.StramAppLauncher.AppFactory; import com.datatorrent.stram.client.StramClientUtils; import com.datatorrent.stram.client.StramClientUtils.ClientRMHelper; import com.datatorrent.stram.codec.LogicalPlanSerializer; import com.datatorrent.stram.plan.logical.LogicalPlan; import com.datatorrent.stram.plan.logical.LogicalPlanConfiguration; import com.datatorrent.stram.plan.logical.requests.AddStreamSinkRequest; import com.datatorrent.stram.plan.logical.requests.CreateOperatorRequest; import com.datatorrent.stram.plan.logical.requests.CreateStreamRequest; import com.datatorrent.stram.plan.logical.requests.LogicalPlanRequest; import com.datatorrent.stram.plan.logical.requests.RemoveOperatorRequest; import com.datatorrent.stram.plan.logical.requests.RemoveStreamRequest; import com.datatorrent.stram.plan.logical.requests.SetOperatorAttributeRequest; import com.datatorrent.stram.plan.logical.requests.SetOperatorPropertyRequest; import com.datatorrent.stram.plan.logical.requests.SetPortAttributeRequest; import com.datatorrent.stram.plan.logical.requests.SetStreamAttributeRequest; import com.datatorrent.stram.security.StramUserLogin; import com.datatorrent.stram.util.JSONSerializationProvider; import com.datatorrent.stram.util.LoggerUtil; import com.datatorrent.stram.util.SecurityUtils; import com.datatorrent.stram.util.VersionInfo; import com.datatorrent.stram.util.WebServicesClient; import com.datatorrent.stram.webapp.OperatorDiscoverer; import com.datatorrent.stram.webapp.StramWebServices; import com.datatorrent.stram.webapp.TypeDiscoverer; import jline.console.ConsoleReader; import jline.console.completer.AggregateCompleter; import jline.console.completer.ArgumentCompleter; import jline.console.completer.Completer; import jline.console.completer.FileNameCompleter; import jline.console.completer.StringsCompleter; import jline.console.history.FileHistory; import jline.console.history.History; import jline.console.history.MemoryHistory; import sun.misc.Signal; import sun.misc.SignalHandler; /** * Provides command line interface for a streaming application on hadoop (yarn) * <p> * * @since 0.3.2 */ @SuppressWarnings("UseOfSystemOutOrSystemErr") public class ApexCli { private static final Logger LOG = LoggerFactory.getLogger(ApexCli.class); private static String CONFIG_EXCLUSIVE = "exclusive"; private static String CONFIG_INCLUSIVE = "inclusive"; private static final String COLOR_RED = "\033[38;5;196m"; private static final String COLOR_YELLOW = "\033[0;93m"; private static final String FORMAT_BOLD = "\033[1m"; private static final String COLOR_RESET = "\033[0m"; private static final String ITALICS = "\033[3m"; private static final String APEX_HIGHLIGHT_COLOR_PROPERTY_NAME = "apex.cli.color.highlight"; private static final String APEX_HIGHLIGHT_COLOR_ENV_VAR_NAME = "APEX_HIGHLIGHT_COLOR"; protected Configuration conf; private FileSystem fs; private StramAgent stramAgent; private YarnClient yarnClient = null; private ApplicationReport currentApp = null; private boolean consolePresent; private String[] commandsToExecute; private final Map<String, CommandSpec> globalCommands = new TreeMap<>(); private final Map<String, CommandSpec> connectedCommands = new TreeMap<>(); private final Map<String, CommandSpec> logicalPlanChangeCommands = new TreeMap<>(); private final Map<String, String> aliases = new HashMap<>(); private final Map<String, List<String>> macros = new HashMap<>(); private boolean changingLogicalPlan = false; private final List<LogicalPlanRequest> logicalPlanRequestQueue = new ArrayList<>(); private FileHistory topLevelHi