dianfu commented on a change in pull request #8675: [FLINK-12716][python] Add an interactive shell for Python Table API URL: https://github.com/apache/flink/pull/8675#discussion_r291990721
########## File path: flink-clients/src/main/java/org/apache/flink/client/cli/PythonShellParser.java ########## @@ -0,0 +1,288 @@ +/* + * 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.flink.client.cli; + +import org.apache.flink.util.Preconditions; + +import org.apache.commons.cli.CommandLine; +import org.apache.commons.cli.DefaultParser; +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 java.util.ArrayList; +import java.util.List; + +/** + * A simple command line parser (based on Apache Commons CLI) that extracts command + * line options.The parser will be used in python shell. + */ +public class PythonShellParser { + private static final Option ADD_JARS_OPTION = new Option("a", "addclasspath", true, + "Specifies additional jars to be used in Flink."); + + private static final Option container = new Option("n", "container", true, + "Number of YARN container to allocate (=Number of Task Managers)"); + + private static final Option jmMemory = new Option("jm", "jobManagerMemory", true, + "Memory for JobManager Container with optional unit (default: MB)"); + + private static final Option name = new Option("nm", "name", true, + "Set a custom name for the application on YARN"); + + private static final Option queue = new Option("qu", "queue", true, + "Specify YARN queue."); + + private static final Option slots = new Option("s", "slots", true, + "Number of slots per TaskManager"); + + private static final Option tmMemory = new Option("tm", "taskManagerMemory", true, + "Memory per TaskManager Container with optional unit (default: MB)"); + + // cluster types + private static final String LOCAL_RUN = "local"; + private static final String REMOTE_RUN = "remote"; + private static final String YARN_RUN = "yarn"; + + // Options that will be used in mini cluster. + private static Options localOptions = new Options(); + + // Options that will be used in remote cluster. + private static Options remoteOptions = new Options(); + + // Options that will be used in yarn cluster. + private static Options yarnOptions = new Options(); + + static { + localOptions.addOption(ADD_JARS_OPTION); + + remoteOptions.addOption(ADD_JARS_OPTION); + + yarnOptions.addOption(container); + yarnOptions.addOption(jmMemory); + yarnOptions.addOption(name); + yarnOptions.addOption(queue); + yarnOptions.addOption(slots); + yarnOptions.addOption(tmMemory); + yarnOptions.addOption(ADD_JARS_OPTION); + } + + public static void main(String[] args) { + if (args.length < 1) { + printError("You should specify cluster type or -h | --help option"); + System.exit(1); + } + String command = args[0]; + List<String> commandOptions = null; + try { + switch (command) { + case LOCAL_RUN: + commandOptions = parseLocal(args); + break; + case REMOTE_RUN: + commandOptions = parseRemote(args); + break; + case YARN_RUN: + commandOptions = parseYarn(args); + break; + case "-h": + case "--help": + printHelp(); + return; + default: + printError(String.format("\"%s\" is not a valid cluster type or -h | --help option.\n", command)); + System.exit(1); + } + for (String option : commandOptions) { + System.out.print(option); + System.out.print('\0'); + } + } catch (Throwable e) { + printError("Error while running the command."); + e.printStackTrace(); + System.exit(1); + } + } + + /** + * Prints the error message and help for the client. + * + * @param msg error message + */ + private static void printError(String msg) { + System.err.println(msg); + System.err.println("Valid cluster type are \"local\", \"remote <hostname> <portnumber>\", \"yarn\"."); + System.err.println(); + System.err.println("Specify the help option (-h or --help) to get help on the command."); + } + + /** + * Prints the help for the client. + */ + private static void printHelp() { + System.out.print("Flink Scala Shell\n"); + System.out.print("Usage: start-scala-shell.sh [local|remote|yarn] [options] <args>...\n"); + System.out.print('\n'); + printLocalHelp(); + printRemoteHelp(); + printYarnHelp(); + System.out.println("-h | --help"); + System.out.println(" Prints this usage text"); + } + + private static void printYarnHelp() { + HelpFormatter formatter = new HelpFormatter(); + formatter.setLeftPadding(5); + formatter.setWidth(80); + System.out.println("Command: yarn [options]"); + System.out.println("Starts Flink scala shell connecting to a yarn cluster"); + formatter.printHelp(" ", yarnOptions); + } + + private static void printRemoteHelp() { + HelpFormatter formatter = new HelpFormatter(); + formatter.setLeftPadding(5); + formatter.setWidth(80); + System.out.println("Command: remote [options] <host> <port>"); + System.out.println("Starts Flink scala shell connecting to a remote cluster"); + System.out.println(" <host>"); + System.out.println(" Remote host name as string"); + System.out.println(" <port>"); + System.out.println(" Remote port as integer"); + System.out.println(); + formatter.printHelp(" ", remoteOptions); + } + + private static void printLocalHelp() { + HelpFormatter formatter = new HelpFormatter(); + formatter.setLeftPadding(5); + formatter.setWidth(80); + System.out.println("Command: local [options]"); + System.out.println("Starts Flink scala shell with a local Flink cluster"); + formatter.printHelp(" ", localOptions); + } + + /** + * Constructs yarn options. The python shell option will add prefix 'y' to align yarn options in {@link CliFrontendParser}. + * + * @param options Options that will be used in `flink run`. + * @param yarnOption Python shell yarn options. + * @param commandLine Parsed Python shell parser options. + */ + private static void constructYarnOption(List<String> options, Option yarnOption, CommandLine commandLine) { + if (commandLine.hasOption(yarnOption.getOpt())) { + options.add("-y" + yarnOption.getOpt()); + options.add(commandLine.getOptionValue(yarnOption.getOpt())); + } + } + + /** + * Parses Python shell yarn options and transfer to yarn options which will be used in `flink run` to + * submit flink job. + * + * @param args Python shell yarn options. + * @return Yarn options usrd in `flink run`. + */ + private static List<String> parseYarn(String[] args) { Review comment: The yarn mode doesn't work: The following exception will be thrown: JAR file does not exist: yarn-cluster ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services
