编辑readme.md

Author: zoudaokoulife

README.md

@@ -1,2 +1,94 @@
 # flinkStreamSQL
-基于开源的flink，对其实时sql进行扩展；主要实现了流与维表的join。
+> * 基于开源的flink，对其实时sql进行扩展   
+>  > 自定义create table 语法（包括输入源表,输出表,维表）
+>  > 自定义create function 语法
+>  > 实现了流与维表的join
+
+
+## 1 快速起步
+### 1.1 运行模式
+
+
+* 单机模式：对应Flink集群的单机模式
+* standalone模式：对应Flink集群的分布式模式
+* yarn模式：对应Flink集群的yarn模式
+
+### 1.2 执行环境
+
+* Java: JDK8及以上
+* Flink集群: 1.4（单机模式不需要安装Flink集群）
+* 操作系统：理论上不限
+
+### 1.3 打包
+
+进入项目根目录，使用maven打包：
+
+```
+mvn clean package -Dmaven.test.skip
+```
+
+打包结束后，项目根目录下会产生plugins目录，plugins目录下存放编译好的数据同步插件包
+
+
+### 1.4 启动
+
+#### 1.4.1 命令行参数选项
+
+* **model**
+	* 描述：执行模式，也就是flink集群的工作模式
+		* local: 本地模式
+		* standalone: 独立部署模式的flink集群
+		* yarn: yarn模式的flink集群
+	* 必选：否
+	* 默认值：local
+	
+* **name**
+	* 描述：flink 任务对应名称。
+	* 必选：是
+	* 默认值：无	
+
+* **sql**
+	* 描述：执行flink sql 的主体语句。
+	* 必选：是
+	* 默认值：无
+	
+* **localSqlPluginPath**
+	* 描述：本地插件根目录地址，也就是打包后产生的plugins目录。
+	* 必选：是
+	* 默认值：无
+	
+* **remoteSqlPluginPath**
+    * 描述：flink执行集群上的插件根目录地址（将打包好的插件存放到各个flink节点上,如果是yarn集群需要存放到所有的nodemanager上）。
+    * 必选：否
+    * 默认值：无
+
+* **addjar**
+    * 描述：扩展jar路径,当前主要是UDF定义的jar；
+    * 必选：否
+    * 默认值：无
+    
+* **confProp**
+    * 描述：一些参数设置
+    * 必选：否
+    * 默认值：无
+    * 可选参数:
+        sql.env.parallelism: 默认并行度设置
+        sql.max.env.parallelism: 最大并行度设置
+        time.characteristic: 可选值[ProcessingTime|IngestionTime|EventTime]
+        sql.checkpoint.interval: 设置了该参数表明开启checkpoint(ms)
+        sql.checkpoint.mode: 可选值[EXACTLY_ONCE|AT_LEAST_ONCE]
+        sql.checkpoint.timeout: 生成checkpoint的超时时间(ms)
+        sql.max.concurrent.checkpoints: 最大并发生成checkpoint数
+        sql.checkpoint.cleanup.mode: 默认是不会将checkpoint存储到外部存储,[true(任务cancel之后会删除外部存储)|false(外部存储需要手动删除)]
+        flinkCheckpointDataURI: 设置checkpoint的外部存储路径,根据实际的需求设定文件路径,hdfs://, file://
+    
+	
+* **flinkconf**
+	* 描述：flink配置文件所在的目录（单机模式下不需要），如/hadoop/flink-1.4.0/conf
+	* 必选：否
+	* 默认值：无
+	
+* **yarnconf**
+	* 描述：Hadoop配置文件（包括hdfs和yarn）所在的目录（单机模式下不需要），如/hadoop/etc/hadoop
+	* 必选：否
+	* 默认值：无

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -119,7 +119,6 @@ public static void main(String[] args) throws Exception {
         Preconditions.checkNotNull(sql, "it requires input parameters sql");
         Preconditions.checkNotNull(name, "it requires input parameters name");
         Preconditions.checkNotNull(localSqlPluginPath, "it requires input parameters localSqlPluginPath");
-        Preconditions.checkNotNull(remoteSqlPluginPath, "it requires input parameters remoteSqlPluginPath");
 
         sql = URLDecoder.decode(sql, Charsets.UTF_8.name());
         SqlParser.setLocalSqlPluginRoot(localSqlPluginPath);
@@ -260,7 +259,7 @@ private static void registerTable(SqlTree sqlTree, StreamExecutionEnvironment en
                 }else{
                     fields += ",PROCTIME.PROCTIME";
                 }
-                //tableEnv.registerDataStream(tableInfo.getName(), adaptStream, fields);
+
                 Table regTable = tableEnv.fromDataStream(adaptStream, fields);
                 tableEnv.registerTable(tableInfo.getName(), regTable);
                 registerTableCache.put(tableInfo.getName(), regTable);

launcher/src/main/java/com/dtstack/flink/sql/launcher/LauncherOptionParser.java

@@ -81,8 +81,11 @@ public LauncherOptionParser(String[] args) {
             String localPlugin = Preconditions.checkNotNull(cl.getOptionValue(OPTION_LOCAL_SQL_PLUGIN_PATH));
             properties.put(OPTION_LOCAL_SQL_PLUGIN_PATH, localPlugin);
 
-            String remotePlugin = Preconditions.checkNotNull(cl.getOptionValue(OPTION_REMOTE_SQL_PLUGIN_PATH));
-            properties.put(OPTION_REMOTE_SQL_PLUGIN_PATH, remotePlugin);
+            String remotePlugin = cl.getOptionValue(OPTION_REMOTE_SQL_PLUGIN_PATH);
+            if(!mode.equalsIgnoreCase(ClusterMode.MODE_LOCAL)){
+                Preconditions.checkNotNull(remotePlugin);
+                properties.put(OPTION_REMOTE_SQL_PLUGIN_PATH, remotePlugin);
+            }
 
             String name = Preconditions.checkNotNull(cl.getOptionValue(OPTION_NAME));
             properties.put(OPTION_NAME, name);