准备工作

• 下载Spark 1.6.2

• 下载StreamingPro

我们假设你下载的StreamingPro包在/tmp目录下。

复制如下模板

{  "esToCsv": {    "desc": "测试",    "strategy": "streaming.core.strategy.SparkStreamingStrategy",    "algorithm": [],    "ref": [],    "compositor": [
      {        "name": "streaming.core.compositor.spark.source.SQLSourceCompositor",        "params": [
          {            "format": "org.elasticsearch.spark.sql",            "path": "索引名称",            "es.nodes": "这里是填写集群地址哈",            "es.mapping.date.rich": "false"
          }
        ]
      },
      {        "name": "streaming.core.compositor.spark.transformation.JSONTableCompositor",        "params": [
          {            "tableName": "table1"
          }
        ]
      },
      {        "name": "streaming.core.compositor.spark.transformation.SQLCompositor",        "params": [
          {            "sql": "select * from table1"
          }
        ]
      },
      {        "name": "streaming.core.compositor.spark.output.SQLOutputCompositor",        "params": [
          {            "format": "com.databricks.spark.csv",            "path": "file:///tmp/csv-table1",            "header": "true",            "inferSchema": "true"
          }
        ]
      }
    ],    "configParams": {
    }
  }
}

假设该文件所在路径是  /tmp/esToCSV.json。

本机运行

cd  $SPARK_HOME./bin/spark-submit   --class streaming.core.StreamingApp 
--master local[2] 
--name test 
/tmp/streamingpro-0.3.2-SNAPSHOT-online-1.6.1.jar    
-streaming.name test    
-streaming.platform spark   
-streaming.job.file.path file:// /tmp/esToCSV.json

在集群运行

cd  $SPARK_HOME./bin/spark-submit   --class streaming.core.StreamingApp 
--master yarn-cluster
--name test 
/tmp/streamingpro-0.3.2-SNAPSHOT-online-1.6.1.jar    
-streaming.name test    
-streaming.platform spark   
-streaming.job.file.path hdfs://clusternameAndPort/tmp/esToCSV.json

作者：祝威廉
链接：https://www.jianshu.com/p/18f949ff1409