---
apiVersion: v1
kind: ConfigMap
metadata:
  labels:
    funktion.fabric8.io/kind: Connector
  name: spark
data:
  deployment.yml: |
    ---
    apiVersion: extensions/v1beta1
    kind: Deployment
    metadata:
      labels:
        funktion.fabric8.io/kind: Subscription
        connector: spark
    spec:
      replicas: 1
      template:
        metadata:
          labels:
            funktion.fabric8.io/kind: Subscription
            connector: spark
        spec:
          containers:
          - image: funktion/connector-spark:${project.version}
            name: connector
  schema.yml: |
    ---
    component:
      kind: component
      scheme: spark
      syntax: spark:endpointType
      title: Apache Spark
      description: The spark component can be used to send RDD or DataFrame jobs to Apache Spark cluster.
      label: bigdata,iot
      deprecated: false
      async: false
      producerOnly: true
      javaType: org.apache.camel.component.spark.SparkComponent
      groupId: org.apache.camel
      artifactId: camel-spark
      version: 2.18.3
    componentProperties:
      rdd:
        kind: property
        type: object
        javaType: org.apache.spark.api.java.JavaRDDLike
        deprecated: false
        secret: false
        description: RDD to compute against.
        order: 0
      rddCallback:
        kind: property
        type: object
        javaType: org.apache.camel.component.spark.RddCallback
        deprecated: false
        secret: false
        description: Function performing action against an RDD.
        order: 1
    properties:
      endpointType:
        kind: path
        group: producer
        required: true
        type: string
        javaType: org.apache.camel.component.spark.EndpointType
        enum:
        - rdd
        - dataframe
        - hive
        deprecated: false
        secret: false
        description: Type of the endpoint (rdd dataframe hive).
        order: 0
      collect:
        kind: parameter
        group: producer
        type: boolean
        javaType: boolean
        deprecated: false
        secret: false
        defaultValue: true
        description: Indicates if results should be collected or counted.
        order: 1
      dataFrame:
        kind: parameter
        group: producer
        type: object
        javaType: org.apache.spark.sql.DataFrame
        deprecated: false
        secret: false
        description: DataFrame to compute against.
        order: 2
      dataFrameCallback:
        kind: parameter
        group: producer
        type: object
        javaType: org.apache.camel.component.spark.DataFrameCallback
        deprecated: false
        secret: false
        description: Function performing action against an DataFrame.
        order: 3
      rdd:
        kind: parameter
        group: producer
        type: object
        javaType: org.apache.spark.api.java.JavaRDDLike
        deprecated: false
        secret: false
        description: RDD to compute against.
        order: 4
      rddCallback:
        kind: parameter
        group: producer
        type: object
        javaType: org.apache.camel.component.spark.RddCallback
        deprecated: false
        secret: false
        description: Function performing action against an RDD.
        order: 5
      synchronous:
        kind: parameter
        group: advanced
        label: advanced
        type: boolean
        javaType: boolean
        deprecated: false
        secret: false
        defaultValue: false
        description: Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported).
        order: 6
  documentation.adoc: |+
    [[ApacheSpark-ApacheSparkcomponent]]
    Apache Spark component
    ~~~~~~~~~~~~~~~~~~~~~~

    INFO: Apache Spark component is available starting from Camel *2.17*.

    This documentation page covers the http://spark.apache.org/[Apache
    Spark] component for the Apache Camel. The main purpose of the Spark
    integration with Camel is to provide a bridge between Camel connectors
    and Spark tasks. In particular Camel connector provides a way to route
    message from various transports, dynamically choose a task to execute,
    use incoming message as input data for that task and finally deliver the
    results of the execution back to the Camel pipeline.

    [[ApacheSpark-Supportedarchitecturalstyles]]
    Supported architectural styles
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

    Spark component can be used as a driver application deployed into an
    application server (or executed as a fat jar).

    image:apache-spark.data/camel_spark_driver.png[image] +

    Spark component can also be submitted as a job directly into the Spark
    cluster.

    image:apache-spark.data/camel_spark_cluster.png[image] +

    While Spark component is primary designed to work as a _long running
    job_ serving as an bridge between Spark cluster and the other endpoints,
    you can also use it as a _fire-once_ short job.   

    [[ApacheSpark-RunningSparkinOSGiservers]]
    Running Spark in OSGi servers
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

    Currently the Spark component doesn't support execution in the OSGi
    container. Spark has been designed to be executed as a fat jar, usually
    submitted as a job to a cluster. For those reasons running Spark in an
    OSGi server is at least challenging and is not support by Camel as well.

    [[ApacheSpark-URIformat]]
    URI format
    ^^^^^^^^^^

    Currently the Spark component supports only producers - it it intended
    to invoke a Spark job and return results. You can call RDD, data frame
    or Hive SQL job.

    *Spark URI format*

    [source,java]
    --------------------------
    spark:{rdd|dataframe|hive}
    --------------------------

    [[ApacheSpark-options]]
    Spark options
    +++++++++++++



    // component options: START
    The Apache Spark component supports 2 options which are listed below.



    {% raw %}
    [width="100%",cols="2,1m,7",options="header"]
    |=======================================================================
    | Name | Java Type | Description
    | rdd | JavaRDDLike | RDD to compute against.
    | rddCallback | RddCallback | Function performing action against an RDD.
    |=======================================================================
    {% endraw %}
    // component options: END




    // endpoint options: START
    The Apache Spark component supports 7 endpoint options which are listed below:

    {% raw %}
    [width="100%",cols="2,1,1m,1m,5",options="header"]
    |=======================================================================
    | Name | Group | Default | Java Type | Description
    | endpointType | producer |  | EndpointType | *Required* Type of the endpoint (rdd dataframe hive).
    | collect | producer | true | boolean | Indicates if results should be collected or counted.
    | dataFrame | producer |  | DataFrame | DataFrame to compute against.
    | dataFrameCallback | producer |  | DataFrameCallback | Function performing action against an DataFrame.
    | rdd | producer |  | JavaRDDLike | RDD to compute against.
    | rddCallback | producer |  | RddCallback | Function performing action against an RDD.
    | synchronous | advanced | false | boolean | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported).
    |=======================================================================
    {% endraw %}
    // endpoint options: END

     
    [[ApacheSpark-RDDjobs]]
    RDD jobs 
    ^^^^^^^^^

    To invoke an RDD job, use the following URI:

    *Spark RDD producer*

    [source,java]
    ------------------------------------------------------
    spark:rdd?rdd=#testFileRdd&rddCallback=#transformation
    ------------------------------------------------------

     Where `rdd` option refers to the name of an RDD instance (subclass of
    `org.apache.spark.api.java.JavaRDDLike`) from a Camel registry, while
    `rddCallback` refers to the implementation
    of `org.apache.camel.component.spark.RddCallback` interface (also from a
    registry). RDD callback provides a single method used to apply incoming
    messages against the given RDD. Results of callback computations are
    saved as a body to an exchange.

    *Spark RDD callback*

    [source,java]
    -------------------------------------------------
    public interface RddCallback<T> {
        T onRdd(JavaRDDLike rdd, Object... payloads);
    }
    -------------------------------------------------

    The following snippet demonstrates how to send message as an input to
    the job and return results:

    *Calling spark job*

    [source,java]
    ------------------------------------------------------------------------------------------------------------------------------
    String pattern = "job input";
    long linesCount = producerTemplate.requestBody("spark:rdd?rdd=#myRdd&rddCallback=#countLinesContaining", pattern, long.class);
    ------------------------------------------------------------------------------------------------------------------------------

    The RDD callback for the snippet above registered as Spring bean could
    look as follows:

    *Spark RDD callback*

    [source,java]
    ------------------------------------------------------------------------
    @Bean
    RddCallback<Long> countLinesContaining() {
        return new RddCallback<Long>() {
            Long onRdd(JavaRDDLike rdd, Object... payloads) {
                String pattern = (String) payloads[0];
                return rdd.filter({line -> line.contains(pattern)}).count();
            }
        }
    }
    ------------------------------------------------------------------------

    The RDD definition in Spring could looks as follows:

    *Spark RDD definition*

    [source,java]
    --------------------------------------------------
    @Bean
    JavaRDDLike myRdd(JavaSparkContext sparkContext) {
      return sparkContext.textFile("testrdd.txt");
    }
    --------------------------------------------------

    [[ApacheSpark-VoidRDDcallbacks]]
    Void RDD callbacks
    ++++++++++++++++++

    If your RDD callback doesn't return any value back to a Camel pipeline,
    you can either return `null` value or use `VoidRddCallback` base class:

    *Spark RDD definition*

    [source,java]
    ------------------------------------------------------------------
    @Bean
    RddCallback<Void> rddCallback() {
      return new VoidRddCallback() {
            @Override
            public void doOnRdd(JavaRDDLike rdd, Object... payloads) {
                rdd.saveAsTextFile(output.getAbsolutePath());
            }
        };
    }
    ------------------------------------------------------------------

    [[ApacheSpark-ConvertingRDDcallbacks]]
    Converting RDD callbacks
    ++++++++++++++++++++++++

    If you know what type of the input data will be sent to the RDD
    callback, you can use `ConvertingRddCallback` and let Camel to
    automatically convert incoming messages before inserting those into the
    callback:

    *Spark RDD definition*

    [source,java]
    ---------------------------------------------------------------------------
    @Bean
    RddCallback<Long> rddCallback(CamelContext context) {
      return new ConvertingRddCallback<Long>(context, int.class, int.class) {
                @Override
                public Long doOnRdd(JavaRDDLike rdd, Object... payloads) {
                    return rdd.count() * (int) payloads[0] * (int) payloads[1];
                }
            };
        };
    }
    ---------------------------------------------------------------------------

    [[ApacheSpark-AnnotatedRDDcallbacks]]
    Annotated RDD callbacks
    +++++++++++++++++++++++

    Probably the easiest way to work with the RDD callbacks is to provide
    class with method marked with `@RddCallback` annotation:

    *Annotated RDD callback definition*

    [source,java]
    -----------------------------------------------------------------------------------------------------
    import static org.apache.camel.component.spark.annotations.AnnotatedRddCallback.annotatedRddCallback;
     
    @Bean
    RddCallback<Long> rddCallback() {
        return annotatedRddCallback(new MyTransformation());
    }
     
    ...
     
    import org.apache.camel.component.spark.annotation.RddCallback;
     
    public class MyTransformation {
     
        @RddCallback
        long countLines(JavaRDD<String> textFile, int first, int second) {
            return textFile.count() * first * second;
        }
     
    }
    -----------------------------------------------------------------------------------------------------

    If you will pass CamelContext to the annotated RDD callback factory
    method, the created callback will be able to convert incoming payloads
    to match the parameters of the annotated method:

    *Body conversions for annotated RDD callbacks*

    [source,java]
    ------------------------------------------------------------------------------------------------------------------------------
    import static org.apache.camel.component.spark.annotations.AnnotatedRddCallback.annotatedRddCallback;
     
    @Bean
    RddCallback<Long> rddCallback(CamelContext camelContext) {
        return annotatedRddCallback(new MyTransformation(), camelContext);
    }
     
    ...

     
    import org.apache.camel.component.spark.annotation.RddCallback;
     
    public class MyTransformation {
     
        @RddCallback
        long countLines(JavaRDD<String> textFile, int first, int second) {
            return textFile.count() * first * second;
        }
     
    }
     
    ...
     
    // Convert String "10" to integer
    long result = producerTemplate.requestBody("spark:rdd?rdd=#rdd&rddCallback=#rddCallback" Arrays.asList(10, "10"), long.class);
    ------------------------------------------------------------------------------------------------------------------------------

     

    [[ApacheSpark-DataFramejobs]]
    DataFrame jobs
    ^^^^^^^^^^^^^^

    Instead of working with RDDs Spark component can work with DataFrames as
    well. 

    To invoke an DataFrame job, use the following URI:

    *Spark RDD producer*

    [source,java]
    --------------------------------------------------------------------------
    spark:dataframe?dataFrame=#testDataFrame&dataFrameCallback=#transformation
    --------------------------------------------------------------------------

     Where `dataFrame` option refers to the name of an DataFrame instance
    (`instance of of org.apache.spark.sql.DataFrame`) from a Camel registry,
    while `dataFrameCallback` refers to the implementation
    of `org.apache.camel.component.spark.DataFrameCallback` interface (also
    from a registry). DataFrame callback provides a single method used to
    apply incoming messages against the given DataFrame. Results of callback
    computations are saved as a body to an exchange.

    *Spark RDD callback*

    [source,java]
    -----------------------------------------------------------
    public interface DataFrameCallback<T> {
        T onDataFrame(DataFrame dataFrame, Object... payloads);
    }
    -----------------------------------------------------------

    The following snippet demonstrates how to send message as an input to a
    job and return results:

    *Calling spark job*

    [source,java]
    -----------------------------------------------------------------------------------------------------------------------------------------
    String model = "Micra";
    long linesCount = producerTemplate.requestBody("spark:dataFrame?dataFrame=#cars&dataFrameCallback=#findCarWithModel", model, long.class);
    -----------------------------------------------------------------------------------------------------------------------------------------

    The DataFrame callback for the snippet above registered as Spring bean
    could look as follows:

    *Spark RDD callback*

    [source,java]
    -------------------------------------------------------------------------------------
    @Bean
    RddCallback<Long> findCarWithModel() {
        return new DataFrameCallback<Long>() {
            @Override
            public Long onDataFrame(DataFrame dataFrame, Object... payloads) {
                String model = (String) payloads[0];
                return dataFrame.where(dataFrame.col("model").eqNullSafe(model)).count();
            }
        };
    }
    -------------------------------------------------------------------------------------

    The DataFrame definition in Spring could looks as follows:

    *Spark RDD definition*

    [source,java]
    ------------------------------------------------------------------------
    @Bean
    DataFrame cars(HiveContext hiveContext) {
        DataFrame jsonCars = hiveContext.read().json("/var/data/cars.json");
        jsonCars.registerTempTable("cars");
        return jsonCars;
    }
    ------------------------------------------------------------------------

    [[ApacheSpark-Hivejobs]]
    Hive jobs
    ^^^^^^^^^

     Instead of working with RDDs or DataFrame Spark component can also
    receive Hive SQL queries as payloads. To send Hive query to Spark
    component, use the following URI:

    *Spark RDD producer*

    [source,java]
    ----------
    spark:hive
    ----------

    The following snippet demonstrates how to send message as an input to a
    job and return results:

    *Calling spark job*

    [source,java]
    ----------------------------------------------------------------------------------------------------
    long carsCount = template.requestBody("spark:hive?collect=false", "SELECT * FROM cars", Long.class);
    List<Row> cars = template.requestBody("spark:hive", "SELECT * FROM cars", List.class);
    ----------------------------------------------------------------------------------------------------

    The table we want to execute query against should be registered in a
    HiveContext before we query it. For example in Spring such registration
    could look as follows:

    *Spark RDD definition*

    [source,java]
    ------------------------------------------------------------------------
    @Bean
    DataFrame cars(HiveContext hiveContext) {
        DataFrame jsonCars = hiveContext.read().json("/var/data/cars.json");
        jsonCars.registerTempTable("cars");
        return jsonCars;
    }
    ------------------------------------------------------------------------

    [[ApacheSpark-SeeAlso]]
    See Also
    ^^^^^^^^

    * link:configuring-camel.html[Configuring Camel]
    * link:component.html[Component]
    * link:endpoint.html[Endpoint]
    * link:getting-started.html[Getting Started]

