Skip to content

streaming-format-converter.md

Latest commit

 

History

History
175 lines (135 loc) · 7.6 KB

streaming-format-converter.md

File metadata and controls

175 lines (135 loc) · 7.6 KB
Raw

Streaming Format Converter

Description

The StreamingFormatConverter application reads any Spark acceptable stream (kafka, socket or file streams like parquet, avro, orc, json or csv), converts it into a DataFrame, optionally partitions it, and saves it in any Spark acceptable data stream format (kafka, socket or file streams like parquet, avro, orc, json or csv).

One of the best features is the easy configuration and the support for output partitioning, by specifying both the partitioning column and the number of partition files per partition folder.

Full support for the parsing of various input files is provided. For more details check the Input Formats section.

Usage Examples

In the examples folder there are multiple examples on how the format converter can be used in real life applications.

The convert-format.sh is an example on how the actual application can be used. Taking out the preparation part, it comes down to the following lines, which can be used in different setups, including the cluster-mode.

spark-submit  -v  \
...
--class org.tupol.spark.tools.StreamingFormatConverter \
--name StreamingFormatConverter \
--files $APPLICATION_CONF \
--jars "$JARS" \
$SPARK_TOOLS_JAR

Notice the following variables:

  • $APPLICATION_CONF needs to point to a file actually called application.conf, somewhere in the local file system.
  • $JARS needs to contains the comma separated list of dependency jars; as of now this list contains the following jars:
    • Scala 2.11: config-1.3.0.jar, scalaz-core_2.11-7.2.26.jar, scala-utils_2.11-0.2.0.jar, spark-utils_2.11-0.4.2.jar
    • Scala 2.12: config-1.3.0.jar, scalaz-core_2.12-7.2.26.jar, scala-utils_2.12-0.2.0.jar, spark-utils_2.12-0.4.2.jar
  • $SPARK_TOOLS_JAR needs to contain the local path to the actual spark-tools jar:
    • Scala 2.11: spark-tools_2.11-0.4.1.jar
    • Scala 2.12: spark-tools_2.12-0.4.1.jar

Configuration Examples Description

Check the README for more details on how to run these examples.

Format Converter Parameters

Common Parameters

  • StreamingFormatConverter.input.format Required

  • StreamingFormatConverter.input.format Required

    • the type of the input file and the corresponding source / parser
    • possible values are:
      • socket
      • kafka
      • file sources: xml, csv, json, parquet, avro, orc and text

  • StreamingFormatConverter.input.schema Optional

    • this is an optional parameter that represents the json Apache Spark schema that should be
      enforced on the input data
    • this schema can be easily obtained from a DataFrame by calling the prettyJson function
    • due to it's complex structure, this parameter can not be passed as a command line argument, but it can only be passed through the application.conf file
    • the schema is applied on read only on file streams; for other streams the developer has to find a way to apply it.
    • StreamingFormatConverter.input.schema.path Optional
      • this is an optional parameter that represents local path or the class path to the json Apache Spark schema that should be enforced on the input data
      • this schema can be easily obtained from a DataFrame by calling the prettyJson function
      • if this parameter is found the schema will be loaded from the given file, otherwise, the schema parameter is tried

  • StreamingFormatConverter.output.format Required

    • the type of the input file and the corresponding source / parser
    • possible values are:
      • kafka
      • file sources: xml, csv, json, parquet, avro, orc and text

  • StreamingFormatConverter.output.trigger Optional

    • type: possible values: "Continuous", "Once", "ProcessingTime"
    • interval: mandatory for "Continuous", "ProcessingTime"

  • StreamingFormatConverter.output.queryName Optional

  • StreamingFormatConverter.output.partition.columns Optional

  • StreamingFormatConverter.output.outputMode Optional

  • StreamingFormatConverter.output.checkpointLocation Optional

The StreamingFormatConverter uses the spark-utils defined IO framework. For more details about defining the data sources please check the DataDource documentation.

For more details about defining the data sinks please check the DataSink documentation.

Input Parameters

File Parameters

Socket Parameters

Warning: Not for production use!

  • options Required
    • host Required
    • port Required
    • includeTimestamp Optional

Kafka Parameters

  • options Required
    • kafka.bootstrap.servers Required
    • assign | subscribe | subscribePattern Required *
    • startingOffsets Optional
    • endingOffsets Optional
    • failOnDataLoss Optional
    • kafkaConsumer.pollTimeoutMs Optional
    • fetchOffset.numRetries Optional
    • fetchOffset.retryIntervalMs Optional
    • maxOffsetsPerTrigger Optional

Output Parameters

Common Parameters

  • format Required
    • the type of the input file and the corresponding source / parser
    • possible values are: xml, csv, json, parquet, avro, orc and text
  • trigger Optional
    • type: possible values: "Continuous", "Once", "ProcessingTime"
    • interval: mandatory for "Continuous", "ProcessingTime"
  • partition.columns Optional
  • outputMode Optional
  • checkpointLocation Optional

File Parameters

Kafka Parameters

  • kafka.bootstrap.servers Required
  • topic Required

References