cobrix

Cobrix - COBOL Data Source for Apache Spark

Pain free Spark/Cobol files integration.

Seamlessly query your COBOL/EBCDIC binary files as Spark Dataframes and streams.

Add mainframe as a source to your data engineering strategy.

Motivation

Among the motivations for this project, it is possible to highlight:

• Lack of expertise in the Cobol ecosystem, which makes it hard to integrate mainframes into data engineering strategies.

• Lack of support from the open-source community to initiatives in this field.

• The overwhelming majority (if not all) of tools to cope with this domain are proprietary.

• Several institutions struggle daily to maintain their legacy mainframes, which prevents them from evolving to more modern approaches to data management.

• Mainframe data can only take part in data science activities through very expensive investments.

Features

• Supports primitive types (although some are “Cobol compiler specific”).

• Supports REDEFINES, OCCURS and DEPENDING ON fields (e.g. unchecked unions and variable-size arrays).

• Supports nested structures and arrays.

• Supports Hadoop (HDFS, S3, …) as well as local file system.

• The COBOL copybooks parser doesn’t have a Spark dependency and can be reused for integrating into other data processing engines.

• Supports reading files compressed in Hadoop-compatible way (gzip, bzip2, etc), but with limited parallelism. Uncompressed files are preferred for performance.

Videos

We have presented Cobrix at DataWorks Summit 2019 and Spark Summit 2019 conferences. The screencasts are available here:

DataWorks Summit 2019 (General Cobrix workflow for hierarchical databases): https://www.youtube.com/watch?v=o_up7X3ZL24

Spark Summit 2019 (More detailed overview of performance optimizations): https://www.youtube.com/watch?v=BOBIdGf3Tm0

Requirements

Linking

You can link against this library in your program at the following coordinates:

Scala 2.11	Scala 2.12	Scala 2.13
groupId: za.co.absa.cobrix artifactId: spark-cobol_2.11 version: 2.11.2	groupId: za.co.absa.cobrix artifactId: spark-cobol_2.12 version: 2.11.2	groupId: za.co.absa.cobrix artifactId: spark-cobol_2.13 version: 2.11.2

Using with Spark shell

This package can be added to Spark using the --packages command line option. For example, to include it when starting the spark shell:

Spark compiled with Scala 2.11

$SPARK_HOME/bin/spark-shell --packages za.co.absa.cobrix:spark-cobol_2.11:2.11.2

Spark compiled with Scala 2.12

$SPARK_HOME/bin/spark-shell --packages za.co.absa.cobrix:spark-cobol_2.12:2.11.2

Spark compiled with Scala 2.13

$SPARK_HOME/bin/spark-shell --packages za.co.absa.cobrix:spark-cobol_2.13:2.11.2

Usage

Quick start

This repository contains several standalone example applications in examples/spark-cobol-app directory. It is a Maven project that contains several examples:

• SparkTypesApp is an example of a very simple mainframe file processing. It is a fixed record length raw data file with a corresponding copybook. The copybook contains examples of various numeric data types Cobrix supports.
• SparkCobolApp is an example of a Spark Job for handling multisegment variable record length mainframe files.
• SparkCodecApp is an example usage of a custom record header parser. This application reads a variable record length file having non-standard RDW headers. In this example RDH header is 5 bytes instead of 4
• SparkCobolHierarchical is an example processing of an EBCDIC multisegment file extracted from a hierarchical database.

The example project can be used as a template for creating Spark Application. Refer to README.md of that project for the detailed guide how to run the examples locally and on a cluster.

When running mvn clean package in examples/spark-cobol-app an uber jar will be created. It can be used to run jobs via spark-submit or spark-shell.

Reading Cobol binary files from Hadoop/local and querying them

1. Create a Spark SQLContext

2. Start a sqlContext.read operation specifying za.co.absa.cobrix.spark.cobol.source as the format

3. Inform the path to the copybook describing the files through ... .option("copybook", "path_to_copybook_file").

   • By default the copybook is expected to be in the default Hadoop filesystem (HDFS, S3, etc).
   • You can specify that a copybook is located in the local file system by adding file:// prefix.
   • For example, you can specify a local file like this .option("copybook", "file:///home/user/data/copybook.cpy").
   • Alternatively, instead of providing a path to a copybook file you can provide the contents of the copybook itself by using .option("copybook_contents", "...copybook contents...").
   • You can store the copybook in the JAR itself at resources section in this case use jar:// prefix, e.g.: .option("copybook", "jar:///copybooks/copybook.cpy").

4. Inform the path to the Hadoop directory containing the files: ... .load("s3a://path_to_directory_containing_the_binary_files")

5. Inform the query you would like to run on the Cobol Dataframe

Below is an example whose full version can be found at za.co.absa.cobrix.spark.cobol.examples.SampleApp and za.co.absa.cobrix.spark.cobol.examples.CobolSparkExample

val sparkBuilder = SparkSession.builder().appName("Example")
val spark = sparkBuilder
  .getOrCreate()

val cobolDataframe = spark
  .read
  .format("cobol")
  .option("copybook", "data/test1_copybook.cob")
  .load("data/test2_data")

cobolDataframe
    .filter("RECORD.ID % 2 = 0") // filter the even values of the nested field 'RECORD_LENGTH'
    .take(10)
    .foreach(v => println(v))

The full example is available here

In some scenarios Spark is unable to find “cobol” data source by it’s short name. In that case you can use the full path to the source class instead: .format("za.co.absa.cobrix.spark.cobol.source")

Cobrix assumes input data is encoded in EBCDIC. You can load ASCII files as well by specifying the following option: .option("encoding", "ascii").

If the input file is a text file (CRLF / LF are used to split records), use .option("is_text", "true").

Multisegment ASCII text files are supported using this option: .option("record_format", "D").

Cobrix has better handling of special characters and partial records using its extension format: .option("record_format", "D2").

Read more on record formats at https://www.ibm.com/docs/en/zos/2.4.0?topic=files-selecting-record-formats-non-vsam-data-sets

Streaming Cobol binary files from a directory

1. Create a Spark StreamContext

2. Import the binary files/stream conversion manager: za.co.absa.spark.cobol.source.streaming.CobolStreamer._

3. Read the binary files contained in the path informed in the creation of the SparkSession as a stream: ... streamingContext.cobolStream()

4. Apply queries on the stream: ... stream.filter("some_filter") ...

5. Start the streaming job.

Below is an example whose full version can be found at za.co.absa.cobrix.spark.cobol.examples.StreamingExample

val spark = SparkSession
  .builder()
  .appName("CobolParser")
  .master("local[2]")
  .config("duration", 2)
  .config("copybook", "path_to_the_copybook")
  .config("path", "path_to_source_directory") // could be both, local or Hadoop (s3://, hdfs://, etc)
  .getOrCreate()          
      
val streamingContext = new StreamingContext(spark.sparkContext, Seconds(3))         
    
import za.co.absa.spark.cobol.source.streaming.CobolStreamer._ // imports the Cobol streams manager

val stream = streamingContext.cobolStream() // streams the binary files into the application    

stream
    .filter(row => row.getAs[Integer]("NUMERIC_FLD") % 2 == 0) // filters the even values of the nested field 'NUMERIC_FLD'
    .print(10)		

streamingContext.start()
streamingContext.awaitTermination()

Using Cobrix from a Spark shell

To query mainframe files interactively using spark-shell you need to provide jar(s) containing Corbrix and it’s dependencies. This can be done either by downloading all the dependencies as separate jars or by creating an uber jar that contains all of the dependencies.

Getting all Cobrix dependencies

Cobrix’s spark-cobol data source depends on the COBOL parser that is a part of Cobrix itself.

The jars that you need to get are:

• spark-cobol_2.12-2.11.2.jar
• cobol-parser_2.12-2.11.2.jar

Versions older than 2.8.0 also need scodec-core_2.12-1.10.3.jar and scodec-bits_2.12-1.1.4.jar.

Versions older than 2.7.1 also need antlr4-runtime-4.8.jar.

After that you can specify these jars in spark-shell command line. Here is an example:

$ spark-shell --packages za.co.absa.cobrix:spark-cobol_2.12:2.11.2
or 
$ spark-shell --master yarn --deploy-mode client --driver-cores 4 --driver-memory 4G --jars spark-cobol_2.12-2.11.2.jar,cobol-parser_2.12-2.11.2.jar

Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Spark context available as 'sc' (master = yarn, app id = application_1535701365011_2721).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.4.5
      /_/

Using Scala version 2.11.8 (OpenJDK 64-Bit Server VM, Java 1.8.0_171)
Type in expressions to have them evaluated.
Type :help for more information.

scala> val df = spark.read.format("cobol").option("copybook", "/data/example1/test3_copybook.cob").load("/data/example1/data")
df: org.apache.spark.sql.DataFrame = [TRANSDATA: struct<CURRENCY: string, SIGNATURE: string ... 4 more fields>]

scala> df.show(false)
+----------------------------------------------------+
|TRANSDATA                                           |
+----------------------------------------------------+
|[GBP,S9276511,Beierbauh.,0123330087,1,89341.00]     |
|[ZAR,S9276511,Etqutsa Inc.,0039003991,1,2634633.00] |
|[USD,S9276511,Beierbauh.,0038903321,0,75.71]        |
|[ZAR,S9276511,Beierbauh.,0123330087,0,215.39]       |
|[ZAR,S9276511,Test Bank,0092317899,1,643.94]        |
|[ZAR,S9276511,Xingzhoug,8822278911,1,998.03]        |
|[USD,S9276511,Beierbauh.,0123330087,1,848.88]       |
|[USD,S9276511,Beierbauh.,0123330087,0,664.11]       |
|[ZAR,S9276511,Beierbauh.,0123330087,1,55262.00]     |
+----------------------------------------------------+
only showing top 20 rows

scala>

Creating an uber jar

Gathering all dependencies manually maybe a tiresome task. A better approach would be to create a jar file that contains all required dependencies (an uber jar aka fat jar).

Creating an uber jar for Cobrix is very easy. Steps to build:

• Install JDK 8
• Install SBT
• Clone Cobrix repository
• Run sbt assembly in the root directory of the repository specifying the Scala and Spark version you want to build for:

  # For Scala 2.11
  sbt -DSPARK_VERSION="2.4.8" ++2.11.12 assembly
  
  # For Scala 2.12
  sbt -DSPARK_VERSION="3.3.4" ++2.12.21 assembly
  sbt -DSPARK_VERSION="3.4.4" ++2.12.21 assembly
  
  # For Scala 2.13
  sbt -DSPARK_VERSION="3.3.4" ++2.13.19 assembly
  sbt -DSPARK_VERSION="3.4.4" ++2.13.19 assembly
  

You can collect the uber jar of spark-cobol either at spark-cobol/target/scala-2.11/ or in spark-cobol/target/scala-2.12/ depending on the Scala version you used. The fat jar will have ‘-bundle’ suffix. You can also download pre-built bundles from https://github.com/AbsaOSS/cobrix/releases/tag/v2.7.3

Then, run spark-shell or spark-submit adding the fat jar as the option.

$ spark-shell --jars spark-cobol_2.12_3.3-2.11.3-SNAPSHOT-bundle.jar

A note for building and running tests on Windows

• java.lang.UnsatisfiedLinkError: org.apache.hadoop.io.nativeio.NativeIO$POSIX.stat is a Hadoop compatibility with Windows issue. The workaround is described here: https://stackoverflow.com/questions/41851066/exception-in-thread-main-java-lang-unsatisfiedlinkerror-org-apache-hadoop-io
• When running assembly with -DSPARK_VERSION=... on Windows, it seems an sbt compatibility with Windows issue: https://stackoverflow.com/questions/59144913/run-sbt-1-2-8-project-with-java-d-options-on-windows You can work around it by using default Spark version for a given Scala version:

  sbt ++2.11.12 assembly
  sbt ++2.12.21 assembly
  sbt ++2.13.18 assembly
  

Detailed Overview of Features

For introductory materials, see the Documentation Index.

Converting mainframe files to Spark DataFrames consists of 3 main steps. Each step has its own options to spark-cobol:

1. Split files by records. Input files need to be split into individual records. Each record is an array of bytes. The split happens depending on the file format. The most important option here is record_format and it can be:
   • F - fixed record length. Each record has the same size. By default the size is determined by the copybook, but you can set this fixed size yourself via record_length.
   • When record_length_field is defined the file has variable record length, and length of each record is defined by that field. You can use expressions to specify adjustments between the value in the field and the actual record size.
   • V - variable record length records with RDW headers. Each record in the file begins with 4 byte headers determining record length. You can specify if RDW headers are big-endian or little-endian via is_rdw_big_endian option.
   • VB - variable record length records with BDW+RDW headers.
   • FB - fixed block record format. The file is split into blocks with BDW headers. Each record in a block has the same length.
   • D - is ASCII text format. Each record is split using the line ending character.
   • If you use a non-standard format, you can implement your own raw record extractor and specify the fully qualified class name via record_extractor option. A record extractor is essentially an iterator that gets an input stream of bytes as input and outputs records as individual arrays of bytes.
2. Decode each record. After files are split by records the copybook is used to decode each record. The copybook defines the structure of the records and how they should be mapped to DataFrame columns. The copybook can be specified via copybook_contents or copybook (path) options.
3. Process segments. Segments define which part of the copybook to apply to each record. Depending on if your file has one or multiple segments Cobrix can map segments to different fields in REDEFINES group. See the section “Automatic segment redefines filtering” and below for more detailed information.

Here is the diagram explaining the process:

Loading several paths

Currently, specifying multiple paths in load() is not supported. Use the following syntax:

    spark
      .read
      .format("cobol")
      .option("copybook_contents", copybook)
      .option("paths", inputPaths.mkString(","))
      .load()

Spark SQL schema extraction

This library also provides convenient methods to extract Spark SQL schemas and Cobol layouts from copybooks.

If you want to extract a Spark SQL schema from a copybook by providing same options you provide to Spark:

// Same options that you use for spark.read.format("cobol").option()
val options = Map("schema_retention_policy" -> "keep_original")

val cobolSchema = CobolSchema.fromSparkOptions(Seq(copybook), options)
val sparkSchema = cobolSchema.getSparkSchema.toString()

println(sparkSchema)

If you want to extract a Spark SQL schema from a copybook using the Cobol parser directly:

import za.co.absa.cobrix.cobol.parser.CopybookParser
import za.co.absa.cobrix.cobol.reader.policies.SchemaRetentionPolicy
import za.co.absa.cobrix.spark.cobol.schema.CobolSchema

val parsedSchema = CopybookParser.parseSimple(copyBookContents)
val cobolSchema = new CobolSchema(parsedSchema, SchemaRetentionPolicy.CollapseRoot, inputFileNameField = "", generateRecordId = false)
val sparkSchema = cobolSchema.getSparkSchema.toString()

println(sparkSchema)

If you want to check the layout of the copybook:

import za.co.absa.cobrix.cobol.parser.CopybookParser

val copyBook = CopybookParser.parseSimple(copyBookContents)
println(copyBook.generateRecordLayoutPositions())

Spark schema metadata

When a copybook is converted to a Spark schema, some information is lost, such as length of string fields or minimum and maximum number of elements in arrays. To preserve this information, Cobrix adds metadata to Spark schema fields. The following metadata is added:

You can access the metadata in the usual way:

// This example returns the maximum length of a string field that is the first field of the copybook
df.schema.fields(0).metadata.getLong("maxLength")

Fixed record length files

Cobrix assumes files has fixed length (F) record format by default. The record length is determined by the length of the record defined by the copybook. But you can specify the record length explicitly:

.option("record_format", "F")
.option("record_length", "250")

Fixed block record formats (FB) are also supported. The support is experimental, if you find any issues, please let us know. When the record format is ‘FB’ you can specify block length or number of records per block. As with ‘F’ if record_length is not specified, it will be determined from the copybook.

Records that have BDWs, but not rdws can be read like this:

.option("record_format", "FB")
.option("record_length", "250")

or simply

.option("record_format", "FB")

Records that have neither BDWs nor RDWs can be read like this:

.option("record_format", "FB")
.option("record_length", "250")
.option("block_length", "500")

or

.option("record_format", "FB")
.option("record_length", "250")
.option("records_per_block", "2")

More on fixed-length record formats: https://www.ibm.com/docs/en/zos/2.3.0?topic=sets-fixed-length-record-formats

Variable length records support

Cobrix supports variable record length files. The only requirement is that such a file should contain a standard 4 byte record header known as Record Descriptor Word (RDW). Such headers are created automatically when a variable record length file is copied from a mainframe. Another type of files are variable blocked length. Such files contain Block Descriptor Word (BDW), as well as Record Descriptor Word (RDW) headers. Any such header can be either big-endian or little-endian. Also, quite often BDW headers need to be adjusted in order to be read properly. See the use cases section below.

To load variable length record file the following option should be specified:

.option("record_format", "V")

To load variable blocked length record file the following option should be specified:

.option("record_format", "VB")

More on record formats: https://www.ibm.com/docs/en/zos/2.3.0?topic=files-selecting-record-formats-non-vsam-data-sets

The space used by the headers (both BDW and RDW) should not be mentioned in the copybook if this option is used. Please refer to the ‘Record headers support’ section below.

If a record of the copybook contains record lengths for each record you can use ‘record_length_field’ like this:

.option("record_format", "F")
.option("record_length_field", "RECORD_LENGTH")

You can use expressions as well:

.option("record_format", "F")
.option("record_length_field", "RECORD_LENGTH + 10")

or

.option("record_format", "F")
.option("record_length_field", "FIELD1 * 10 + 200")

If the record field contains a string that can be mapped to a record size, you can add the mapping as a JSON:

.option("record_format", "F")
.option("record_length_field", "FIELD_STR")
.option("record_length_map", """{"SEG1":100,"SEG2":200}""")  

You can specify the default record size by defining the key “_”:

.option("record_format", "F")
.option("record_length_field", "FIELD_STR")
.option("record_length_map", """{"SEG1":100,"SEG2":200,"_":100}""")  

Use cases for various variable length formats

In order to understand the file format it is often sufficient to look at the first 4 bytes of the file (un case of RDW only files), or the first 8 bytes of a file + lookup the offset of the block (in case of BDW + RDW)

V header examples (have only RDW headers)

In order to determine if an RDW is a big- or little-endian, take a look at the first 4 bytes. If the first 2 bytes are zeros, it’s a little-endian RDW header, otherwise it is a big-endian RDW header.

VB header examples (have both BDW and RDW headers)

It is harder to determine if a BDW header is big- or little-endian since BDW header bytes can be all non-zero. But for VB format RDWs follow BDWs and endiness. You can determine the endiness from an RDW, and use the same option for BDW.

Variable-size OCCURS

When copybooks contain the OCCURS DEPENDING ON clause, different mainframe systems may export such fields in different ways. By default, Cobrix assumes that fields with OCCURS always occupy the maximum possible space in the record, regardless of the actual number of elements present in the array. This means that the record size does not change based on the actual element count — the array is treated as if it always contains the maximum number of elements. However, if your data uses true variable-size OCCURS fields, you must explicitly select one of the supported options for handling variable-size arrays.

Cobrix supports the following modes:

• Maximum size (default). Arrays always occupy space for the maximum number of elements defined in the copybook. The record size remains constant, even if fewer elements are present.

  .option("variable_size_occurs", "max_size") 
  // Equivalent to "false" in older Cobrix versions; retained for backward compatibility.
  

• Shift records. If the number of elements in an array is less than the maximum, the field following the array starts immediately after the last actual element. All subsequent fields are shifted forward, and the record becomes variable in size.

  .option("variable_size_occurs", "shift_record")
  // Equivalent to "true" in older Cobrix versions; retained for backward compatibility.
  

• Pad records. This mode is similar to Shift record, in that fields following the array begin immediately after the last actual element. However, the record is then padded so that its total size matches the size it would have if all arrays contained the maximum number of elements.

  .option("variable_size_occurs", "pad_record")
  

Schema collapsing

Mainframe data often contain only one root GROUP. In such cases such a GROUP can be considered something similar to XML rowtag. Cobrix allows either to collapse or to retain the GROUP. To turn this on use the following option:

.option("schema_retention_policy", "collapse_root")

or

.option("schema_retention_policy", "keep_original")

Let’s look at an example. Let’s say we have a copybook that looks like this:

       01  RECORD.
           05  ID                        PIC S9(4)  COMP.
           05  COMPANY.
               10  SHORT-NAME            PIC X(10).
               10  COMPANY-ID-NUM        PIC 9(5) COMP-3.

When “schema_retention_policy” is set to “collapse_root” (default), the root group will be collapsed and the schema will look like this (note the RECORD field is not part of the schema):

root
 |-- ID: integer (nullable = true)
 |-- COMPANY: struct (nullable = true)
 |    |-- SHORT_NAME: string (nullable = true)
 |    |-- COMPANY_ID_NUM: integer (nullable = true)

But when “schema_retention_policy” is set to “keep_original”, the schema will look like this (note the RECORD field is part of the schema):

root
 |-- RECORD: struct (nullable = true)
 |    |-- ID: integer (nullable = true)
 |    |-- COMPANY: struct (nullable = true)
 |    |    |-- SHORT_NAME: string (nullable = true)
 |    |    |-- COMPANY_ID_NUM: integer (nullable = true)

You can experiment with this feature using built-in example in za.co.absa.cobrix.spark.cobol.examples.CobolSparkExample

Record Id fields generation

For data that has record order dependency generation of “File_Id”, “Record_Id”, and “Record_Byte_Length” fields is supported. The values of the File_Id column will be unique for each file when a directory is specified as the source for data. The values of the Record_Id column will be unique and sequential record identifiers within the file.

Turn this feature on use

.option("generate_record_id", true)

The following fields will be added to the top of the schema:

root
 |-- File_Id: integer (nullable = false)
 |-- Record_Id: long (nullable = false)
 |-- Record_Byte_Length: integer (nullable = false)

You can use this option to generate raw bytes of each record as a binary field:

.option("generate_record_bytes", "true")

The following fields will be added to the top of the schema:

root
 |-- Record_Bytes: binary (nullable = false)

You can generate _corrupt_fields that will contain original binary values of fields Cobrix was unable to decode:

.option("generate_corrupt_fields", "true")

Locality optimization for variable-length records parsing

Variable-length records depend on headers to have their length calculated, which makes it hard to achieve parallelism while parsing.

Cobrix strives to overcome this drawback by performing a two-stages parsing. The first stage traverses the records retrieving their lengths and offsets into structures called indexes. Then, the indexes are distributed across the cluster, which allows for parallel variable-length records parsing.

However effective, this strategy may also suffer from excessive shuffling, since indexes may be sent to executors far from the actual data.

The latter issue is overcome by extracting the preferred locations for each index directly from HDFS/S3/…, and then passing those locations to Spark during the creation of the RDD that distributes the indexes.

When processing large collections, the overhead of collecting the locations is offset by the benefits of locality, thus, this feature is enabled by default, but can be disabled by the configuration below:

.option("improve_locality", false)

Workload optimization for variable-length records parsing

This feature works only for HDFS, not for any other of Hadoop filesystems.

When dealing with variable-length records, Cobrix strives to maximize locality by identifying the preferred locations in the cluster to parse each record, i.e. the nodes where the record resides.

This feature is implemented by querying HDFS about the locations of the blocks containing each record and instructing Spark to create the partition for that record in one of those locations.

However, sometimes, new nodes can be added to the cluster after the Cobol file is stored, in which case those nodes would be ignored when processing the file since they do not contain any record.

To overcome this issue, Cobrix also strives to re-balance the records among the new nodes at parsing time, as an attempt to maximize the utilization of the cluster. This is done through identifying the busiest nodes and sharing part of their burden with the new ones.

Since this is not an issue present in most cluster configurations, this feature is disabled by default, and can be enabled from the configuration below:

.option("optimize_allocation", true)

If however the option improve_locality is disabled, this option will also be disabled regardless of the value in optimize_allocation.

Record headers support

As you may already know a file in the mainframe world does not mean the same as in the PC world. On PCs we think of a file as a stream of bytes that we can open, read/write and close. On mainframes a file can be a set of records that we can query. Record is a blob of bytes, can have different size. Mainframe’s ‘filesystem’ handles the mapping between logical records and physical location of data.

Details are available at this Wikipedia article (look for MVS filesystem).

So usually a file cannot simply be ‘copied’ from a mainframe. When files are transferred using tools like XCOM each record is prepended with an additional record header or RDW. This header allows readers of a file in PC to restore the ‘set of records’ nature of the file.

Mainframe files coming from IMS and copied through specialized tools contain records (the payload) having schema of DBs copybook warped with DB export tool headers wrapped with record headers. Like this:

RECORD_HEADERS ( TOOL_HEADERS ( PAYLOAD ) )

Similar to Internet’s TCP protocol IP_HEADERS ( TCP_HEADERS ( PAYLOAD ) ).

TOOL_HEADERS are application dependent. Often it contains the length of the payload. But this length is sometime not very reliable. RECORD_HEADERS contain the record length (including TOOL_HEADERS length) and are proved to be reliable.

For fixed record length files record headers can be ignored since we already know the record length. But for variable record length files and for multisegment files record headers can be considered the most reliable single point of truth about record length.

You can instruct the reader to use 4 byte record headers to extract records from a mainframe file.

.option("record_format", "V")

This is very helpful for multisegment files when segments have different lengths. Since each segment has it’s own copybook it is very convenient to extract segments one by one by combining record_format = V option with segment filter option.

.option("segment_field", "SEG-ID")
.option("segment_filter", "1122334")

In this example it is expected that the copybook has a field with the name ‘SEG-ID’. The data source will read all segments, but will parse only ones that have SEG-ID = "1122334".

If you want to parse multiple segments, set the option ‘segment_filter’ to a comma separated list of the segment values. For example:

.option("segment_field", "SEG-ID")
.option("segment_filter", "1122334,1122335")

will only parse the records with SEG-ID = "1122334" OR SEG-ID = "1122335"

Custom record extractors

Custom record extractors can be used for customizing splitting of input files into a set of records. Cobrix supports text files, fixed length binary files and binary files with RDWs. If your input file is not in one of the supported formats you can implement a custom record extractor interface and provide it to spark-cobol as a option:

.option("record_extractor", "com.example.record.header.parser")

A custom record extractor needs to be a class having this precise constructor signature:

class TextRecordExtractor(ctx: RawRecordExtractorParameters) extends Serializable with RawRecordExtractor {
                             // Your implementation
                          }

A record extractor is essentially iterator of records. Each returned record is an array of bytes parsable by the copybook.

A record extractor is invoked two times. First, it is invoked at the beginning each file to go thought the file and create a sparse index. The second time it is invoked by parallel processes starting from different records in the file. The starting record number is provided in constructor. The starting file offset is available from inputStream.

RawRecordContext consists of the following fields that the custom record extractor will get from Cobrix in runtime:

• startingRecordNumber - A record number the input stream is pointing to.
• inputStream - The input stream of bytes of the input file.
• copybook - The parsed copybook of the input stream.
• additionalInfo - An arbitrary info that can be passed as an option (see below).

If your record extractor needs additional information in order to extract records properly, you can provide an arbitrary additional info to the record extracted at runtime by specifying this option:

Take a look at CustomRecordExtractorMock inside spark-cobol project to see how a custom record extractor can be built.

.option("re_additional_info", "some info")

Custom record header parsers (deprecated)

Custom record header parsers are deprecated. Use custom record extractors instead. They are more flexible and easier to use.

If your variable length file does not have RDW headers, but has fields that can be used for determining record lengths you can provide a custom record header parser that takes starting bytes of each record and returns record lengths. In order to do that you need to create a class inheriting RecordHeaderParser and Serializable traits and provide a fully qualified class name to the following option:

.option("record_header_parser", "com.example.record.header.parser")

RDDs

Cobrix provides helper methods to convert RDD[String] or RDD[Array[Byte]] to DataFrame using a copybook. This can be used if you want to use a custom logic to split the input file into records as either ASCII strings or arrays of bytes, and then parse each record using a copybook.

An example of RDD[Array[Byte]]:

import za.co.absa.cobrix.spark.cobol.Cobrix

val rdd = ???
val df = Cobrix.fromRdd
    .copybookContents(copybook)
    .option("encoding", "ebcdic") // any supported option 
    .load(rdd)

An example of ASCII Strings RDD[String]:

import za.co.absa.cobrix.spark.cobol.Cobrix

val rdd = ???
val df = Cobrix.fromRdd
    .copybookContents(copybook)
    .option("variable_size_occurs", "true") // any supported option 
    .loadText(rdd)

When converting from an RDD some of the options like record_format or generate_record_id cannot be used since the data is assumed to be already split by records and the information about file names and relative order of records is not available.

Options validation

You can validate options to spark-cobol options without reading ro writing data files. This is useful for quick validation separately from the routine that processes the files.

An example:

import za.co.absa.cobrix.spark.cobol.Cobrix

// Validation options when reading data
Cobrix.validateOptions
  .option("copybook_contents", copybook)
  .option("some_option", "some_value")
  .validateLoad(inpudDataPath)

// Validation options when writing data
Cobrix.validateOptions
  .option("copybook_contents", copybookContents)
  .validateSave(df)

EBCDIC code pages

The following code pages are supported:

• common - (default) EBCDIC common characters
• common_extended - EBCDIC common characters with special characters extension
• cp037 - IBM EBCDIC US-Canada
• cp037_extended - IBM EBCDIC US-Canada with special characters extension
• cp300 - IBM EBCDIC Japanese Extended (2 byte code page)
• cp838 - IBM EBCDIC Thailand
• cp870 - IBM EBCDIC Multilingual Latin-2
• cp875 - IBM EBCDIC Greek
• cp1025 - IBM EBCDIC Multilingual Cyrillic
• cp1047 - IBM EBCDIC Latin-1/Open System
• cp1364 - (experimental support) IBM EBCDIC Korean (2 byte code page)
• cp1388 - (experimental support) IBM EBCDIC Simplified Chinese (2 byte code page)

By default, Cobrix uses common EBCDIC code page which contains only basic latin characters, numbers, and punctuation. You can specify the code page to use for all string fields by setting the ebcdic_code_page option to one of the following values:

.option("ebcdic_code_page", "cp037")

For multi-codepage files, you can specify the code page to use for each field by setting the field_code_page:<code page> option

.option("ebcdic_code_page", "cp037")
.option("field_code_page:cp1256" -> "FIELD1")
.option("field_code_page:us-ascii" -> "FIELD-2, FIELD_3")

Reading ASCII text file

Cobrix is primarily designed to read binary files, but you can directly use some internal functions to read ASCII text files. In ASCII text files, records are separated with newlines.

Working example 1:

    // The recommended way
    val df = spark
      .read
      .format("cobol")
      .option("copybook_contents", copybook)
      .option("ascii_charset", "ISO-8859-1") // You can choose a charset, UTF-8 is used by default
      .option("record_format", "D")
      .load(tmpFileName)

Working example 2 - Using RDDs and helper methods:

    // This is the way if you have data converted to an RDD[String] already.
    // You have full control on reading the input data records and converting them to `java.lang.String`.
    val df = Cobrix.fromRdd
        .copybookContents(copybook)
        .option("variable_size_occurs", "true") // any supported option 
        .loadText(rdd)

Working example 3 - Using RDDs and record parsers directly:

    // This is the most verbose way - creating dataframes from RDDs. But it gives full control on how text files are
    // processed before parsing actual records
    val spark = SparkSession
      .builder()
      .appName("Spark-Cobol ASCII text file")
      .master("local[*]")
      .getOrCreate()

    val copybook =
      """       01  COMPANY-DETAILS.
        |            05  SEGMENT-ID		PIC 9(1).
        |            05  STATIC-DETAILS.
        |               10  NAME      	PIC X(2).
        |
        |            05  CONTACTS REDEFINES STATIC-DETAILS.
        |               10  PERSON    	PIC X(3).
      """.stripMargin

    val parsedCopybook = CopybookParser.parse(copybook, dataEnncoding = ASCII, stringTrimmingPolicy = StringTrimmingPolicy.TrimNone)
    val cobolSchema = new CobolSchema(parsedCopybook, SchemaRetentionPolicy.CollapseRoot, "", false)
    val sparkSchema = cobolSchema.getSparkSchema

    val rddText = spark.sparkContext.textFile("src/main/resources/mini.txt")

    val recordHandler = new RowHandler()

    val rddRow = rddText
      .filter(str => str.length > 0)
      .map(str => {
        val record = RecordExtractors.extractRecord[GenericRow](parsedCopybook.ast,
          str.getBytes(),
          0,
          SchemaRetentionPolicy.CollapseRoot, handler = recordHandler)
        Row.fromSeq(record)
      })

    val dfOut = spark.createDataFrame(rddRow, sparkSchema)

    dfOut.printSchema()
    dfOut.show()

Corresponding data sample in mini.txt:

1BB
2CCC

Output:

root
 |-- SEGMENT_ID: integer (nullable = true)
 |-- STATIC_DETAILS: struct (nullable = true)
 |    |-- NAME: string (nullable = true)
 |-- CONTACTS: struct (nullable = true)
 |    |-- PERSON: string (nullable = true)

 ...

 +----------+--------------+--------+
 |SEGMENT_ID|STATIC_DETAILS|CONTACTS|
 +----------+--------------+--------+
 |         1|          [BB]|  [null]|
 |         2|          [CC]|   [CCC]|
 +----------+--------------+--------+

There, Cobrix loaded all redefines for every record. Each record contains data from all of the segments. But only one redefine is valid for every segment. Filtering is described in the following section.

Automatic segment redefines filtering

When reading a multisegment file you can use Spark to clean up redefines that do not match segment ids. Cobrix will parse every redefined field for each segment. To increase performance you can specify which redefine corresponds to which segment id. This way Cobrix will parse only relevant segment redefined fields and leave the rest of the redefined fields null.

  .option("redefine-segment-id-map:0", "REDEFINED_FIELD1 => SegmentId1,SegmentId2,...")
  .option("redefine-segment-id-map:1", "REDEFINED_FIELD2 => SegmentId10,SegmentId11,...")

For the above example the load options will look like this (last 2 options):

val df = spark
  .read
  .format("cobol")
  .option("copybook_contents", copybook)
  .option("record_format", "V")
  .option("segment_field", "SEGMENT_ID")
  .option("segment_id_level0", "C")
  .option("segment_id_level1", "P")
  .option("redefine_segment_id_map:0", "STATIC-DETAILS => C")
  .option("redefine_segment_id_map:1", "CONTACTS => P")
  .load("examples/multisegment_data/COMP.DETAILS.SEP30.DATA.dat")

The filtered data will look like this:

df.show(10)
+----------+----------+--------------------+--------------------+
|SEGMENT_ID|COMPANY_ID|      STATIC_DETAILS|            CONTACTS|
+----------+----------+--------------------+--------------------+
|         C|9377942526|[Joan Q & Z,10 Sa...|                    |
|         P|9377942526|                    |[+(277) 944 44 55...|
|         C|3483483977|[Robotrd Inc.,2 P...|                    |
|         P|3483483977|                    |[+(174) 970 97 54...|
|         P|3483483977|                    |[+(848) 832 61 68...|
|         P|3483483977|                    |[+(455) 184 13 39...|
|         C|7540764401|[Eqartion Inc.,87...|                    |
|         C|4413124035|[Xingzhoug,74 Qin...|                    |
|         C|9546291887|[ZjkLPj,5574, Tok...|                    |
|         P|9546291887|                    |[+(300) 252 33 17...|
+----------+----------+--------------------+--------------------+

In the above example invalid fields became null and the parsing is done faster because Cobrix does not need to process every redefine for each record.

Automatic filtering of arbitrary redefines

Arbitrary redefines can be resolved using rule expressions. This doesn’t have to be segment redefines, just any redefines. For example, for a copybook that looks like this:

        01  COMPANY-DETAILS.
************** RECORD-TYPE CAN BE 'C' for company, and 'P' or 'E' for person.        
            05  RECORD-TYPE          PIC X(1).
            05  COMPANY-ID           PIC X(10).
            05  COMPANY.
               10  COMPANY-NAME      PIC X(15).
               10  ADDRESS           PIC X(25).
            05  PERSON REDEFINES COMPANY.
               10  FIRST-NAME        PIC X(30).
               10  LAST-NAME         PIC X(30).

The syntax is as follows:

  .option("redefine_rule:1", "COMPANY => RECORD_TYPE = 'C'")
  .option("redefine_rule:2", "PERSON => in(RECORD_TYPE, 'P', 'E')")

For the above example the load options will look like this (last 2 options):

val df = spark
  .read
  .format("cobol")
  .option("copybook_contents", copybook)
  .option("record_format", "V")
  .option("redefine_rule:1", "COMPANY => RECORD_TYPE = 'C'")
  .option("redefine_rule:2", "PERSON => in(RECORD_TYPE, 'P', 'E')")
  .load("examples/multisegment_data/COMP.DETAILS.SEP30.DATA.dat")

The filtered data will look like this:

df.show(10)
+-----------+----------+--------------------+--------------------+
|RECORD_TYPE|COMPANY_ID|             COMPANY|              PERSON|
+-----------+----------+--------------------+--------------------+
|          C|9377942526|[Joan Q & Z,10 Sa...|                    |
|          P|9377942526|                    |       [John, Smith]|
|          C|3483483977|[Robotrd Inc.,2 P...|                    |
|          E|3483483977|                    |      [Jane, Wanson]|
|          E|3483483977|                    |      [Alex,Johnson]|
+-----------+----------+--------------------+--------------------+

Notes

• Variable names in rule expressions are case-sensitive.
• Variable names are required to be used after column sanitization (e.g. replacement of special characters with underscores), otherwise expression F-A = 1 is ambiguous since it is not clear if F-A is a variable name or an expression of subtraction. In this case the variable name should be F_A and the expression should be F_A = 1.
• You can only reference variables that go before the redefine field. This is because record decoding is forward only.
• Use only field names themselves, not full paths, e.g. COMPANY instead of RECORD.DETAILS.COMPANY .
• Only integral numeric literals are supported. Decimals are not supported.
• The expression should return a boolean. For example:
  • RECORD_TYPE = 'C' is valid since it returns true for company records and false for person records.
  • in(RECORD_TYPE, 'P', 'E') is valid since it returns true for person records and false for company records.
  • COMPANY_ID > 1000 is valid since it returns true for records with company id greater than 1000 and false otherwise.

Expressions supported

• Comparison operators: =, !=, >, <, >=, <=.
• Boolean logic: && (and), || (or), ! (not).
• Integral literals: 123, 0, -456 are valid, but 123.45 or -123.45 are not valid.
• String literals: 'abc', 'abc', '123' are valid. Always use single quote character.
• Boolean literals: true, false.
• Null literal: null. For example: RECORD_TYPE = 'C' || RECORD_TYPE = null is valid.
• Functions:
  • in() (one of from the list), for example: in(RECORD_TYPE, 'P', 'E') is valid since it returns true for person records and false for company records.
  • if() (conditional function with 3 arguments), for example: if(RECORD_TYPE = 'C', true, false) is valid.

Group Filler dropping

A FILLER is an anonymous field that is usually used for reserving space for new fields in a fixed record length data. Or it is used to remove a field from a copybook without affecting compatibility.

      05  COMPANY.
          10  NAME      PIC X(15).
          10  FILLER    PIC X(5).
          10  ADDRESS   PIC X(25).
          10  FILLER    PIC X(125).

Such fields are dropped when imported into a Spark data frame by Cobrix. Some copybooks, however, have FILLER groups that contain non-filler fields. For example,

      05  FILLER.
          10  NAME      PIC X(15).
          10  ADDRESS   PIC X(25).
      05  FILLER.
          10  AMOUNT    PIC 9(10)V96.
          10  COMMENT   PIC X(40).

By default Cobrix will retain such fields, but will rename each such filler to a unique name so each each individual struct can be specified unambiguously. For example, in this case the filler groups will be renamed to FILLER_1 and FILLER_2. You can change this behaviour if you would like to drop such filler groups by providing this option:

.option("drop_group_fillers", "true")

In order to retain value FILLERs (e.g. non-group FILLERs) as well, use this option:

.option("drop_value_fillers", "false")

Reading hierarchical data sets

        01  COMPANY-DETAILS.
            05  SEGMENT-ID        PIC X(5).
            05  COMPANY-ID        PIC X(10).
            05  STATIC-DETAILS.
               10  COMPANY-NAME      PIC X(15).
               10  ADDRESS           PIC X(25).
               10  TAXPAYER.
                  15  TAXPAYER-TYPE  PIC X(1).
                  15  TAXPAYER-STR   PIC X(8).
                  15  TAXPAYER-NUM  REDEFINES TAXPAYER-STR
                                     PIC 9(8) COMP.

            05  CONTACTS REDEFINES STATIC-DETAILS.
               10  PHONE-NUMBER      PIC X(17).
               10  CONTACT-PERSON    PIC X(28).

val df = spark
  .read
  .format("cobol")
  .option("copybook", "/path/to/thecopybook")
  .option("record_format", "V")
  .load("examples/multisegment_data")

df.printSchema()
root
 |-- SEGMENT_ID: string (nullable = true)
 |-- COMPANY_ID: string (nullable = true)
 |-- STATIC_DETAILS: struct (nullable = true)
 |    |-- COMPANY_NAME: string (nullable = true)
 |    |-- ADDRESS: string (nullable = true)
 |    |-- TAXPAYER: struct (nullable = true)
 |    |    |-- TAXPAYER_TYPE: string (nullable = true)
 |    |    |-- TAXPAYER_STR: string (nullable = true)
 |    |    |-- TAXPAYER_NUM: integer (nullable = true)
 |-- CONTACTS: struct (nullable = true)
 |    |-- PHONE_NUMBER: string (nullable = true)
 |    |-- CONTACT_PERSON: string (nullable = true)

df.show(10)
+----------+----------+--------------------+--------------------+
|SEGMENT_ID|COMPANY_ID|      STATIC_DETAILS|            CONTACTS|
+----------+----------+--------------------+--------------------+
|         C|9377942526|[Joan Q & Z,10 Sa...|[Joan Q & Z     1...|
|         P|9377942526|[+(277) 944 44 5,...|[+(277) 944 44 55...|
|         C|3483483977|[Robotrd Inc.,2 P...|[Robotrd Inc.   2...|
|         P|3483483977|[+(174) 970 97 5,...|[+(174) 970 97 54...|
|         P|3483483977|[+(848) 832 61 6,...|[+(848) 832 61 68...|
|         P|3483483977|[+(455) 184 13 3,...|[+(455) 184 13 39...|
|         C|7540764401|[Eqartion Inc.,87...|[Eqartion Inc.  8...|
|         C|4413124035|[Xingzhoug,74 Qin...|[Xingzhoug      7...|
|         C|9546291887|[ZjkLPj,5574, Tok...|[ZjkLPj         5...|
|         P|9546291887|[+(300) 252 33 1,...|[+(300) 252 33 17...|
+----------+----------+--------------------+--------------------+

Automatic reconstruction of hierarchical record structure

val df = spark
  .read
  .format("cobol")
  .option("copybook", "/path/to/thecopybook")
  .option("record_format", "V")

  // Specifies a field containing a segment id
  .option("segment_field", "SEGMENT_ID")
  
  // Specifies a mapping between segment ids and segment redefine fields
  .option("redefine_segment_id_map:1", "STATIC-DETAILS => C")
  .option("redefine-segment-id-map:2", "CONTACTS => P")
  
  // Specifies a parent-child relationship
  .option("segment-children:1", "STATIC-DETAILS => CONTACTS")
  
  .load("examples/multisegment_data")

scala> df.printSchema()

root
 |-- SEGMENT_ID: string (nullable = true)
 |-- COMPANY_ID: string (nullable = true)
 |-- STATIC_DETAILS: struct (nullable = true)
 |    |-- COMPANY_NAME: string (nullable = true)
 |    |-- ADDRESS: string (nullable = true)
 |    |-- TAXPAYER: struct (nullable = true)
 |    |    |-- TAXPAYER_TYPE: string (nullable = true)
 |    |    |-- TAXPAYER_STR: string (nullable = true)
 |    |    |-- TAXPAYER_NUM: integer (nullable = true)
 |    |-- CONTACTS: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- PHONE_NUMBER: string (nullable = true)
 |    |    |    |-- CONTACT_PERSON: string (nullable = true)

scala> import za.co.absa.cobrix.spark.cobol.utils.SparkUtils

scala> println(SparkUtils.prettyJSON(df.toJSON.take(1).mkString("[", ", ", "]")))
{
  "SEGMENT_ID" : "C",
  "COMPANY_ID" : "9377942526",
  "STATIC_DETAILS" : {
    "COMPANY_NAME" : "Joan Q & Z",
    "ADDRESS" : "10 Sandton, Johannesburg",
    "TAXPAYER" : {
      "TAXPAYER_TYPE" : "A",
      "TAXPAYER_STR" : "92714306",
      "TAXPAYER_NUM" : 959592241
    },
    "CONTACTS" : [ {
      "PHONE_NUMBER" : "+(174) 970 97 54",
      "CONTACT_PERSON" : "Tyesha Debow"
    }, {
      "PHONE_NUMBER" : "+(848) 832 61 68",
      "CONTACT_PERSON" : "Mindy Celestin"
    }, {
      "PHONE_NUMBER" : "+(455) 184 13 39",
      "CONTACT_PERSON" : "Mabelle Winburn"
    } ]
  }
}

import spark.implicits._

val dfCompanies = df
  // Filtering the first segment by segment id
  .filter($"SEGMENT_ID"==="C")
  // Selecting fields that are only available in the first segment
  .select($"COMPANY_ID", $"STATIC_DETAILS.COMPANY_NAME", $"STATIC_DETAILS.ADDRESS",
  // Resolving the taxpayer redefine
    when($"STATIC_DETAILS.TAXPAYER.TAXPAYER_TYPE" === "A", $"STATIC_DETAILS.TAXPAYER.TAXPAYER_STR")
      .otherwise($"STATIC_DETAILS.TAXPAYER.TAXPAYER_NUM").cast(StringType).as("TAXPAYER"))

dfCompanies.show(10, truncate = false)
+----------+-------------+-------------------------+--------+
|COMPANY_ID|COMPANY_NAME |ADDRESS                  |TAXPAYER|
+----------+-------------+-------------------------+--------+
|9377942526|Joan Q & Z   |10 Sandton, Johannesburg |92714306|
|3483483977|Robotrd Inc. |2 Park ave., Johannesburg|31195396|
|7540764401|Eqartion Inc.|871A Forest ave., Toronto|87432264|
|4413124035|Xingzhoug    |74 Qing ave., Beijing    |50803302|
|9546291887|ZjkLPj       |5574, Tokyo              |73538919|
|9168453994|Test Bank    |1 Garden str., London    |82573513|
|4225784815|ZjkLPj       |5574, Tokyo              |96136195|
|8463159728|Xingzhoug    |74 Qing ave., Beijing    |17785468|
|8180356010|Eqartion Inc.|871A Forest ave., Toronto|79054306|
|7107728116|Xingzhoug    |74 Qing ave., Beijing    |70899995|
+----------+-------------+-------------------------+--------+

    val dfContacts = df
      // Filtering the second segment by segment id
      .filter($"SEGMENT_ID"==="P")
      // Selecting the fields only valid for the second segment
      .select($"COMPANY_ID", $"CONTACTS.CONTACT_PERSON", $"CONTACTS.PHONE_NUMBER")

dfContacts.show(10, truncate = false)
+----------+--------------------+----------------+
|COMPANY_ID|CONTACT_PERSON      |PHONE_NUMBER    |
+----------+--------------------+----------------+
|9377942526|Janiece Newcombe    |+(277) 944 44 55|
|3483483977|Tyesha Debow        |+(174) 970 97 54|
|3483483977|Mindy Celestin      |+(848) 832 61 68|
|3483483977|Mabelle Winburn     |+(455) 184 13 39|
|9546291887|Carrie Celestin     |+(300) 252 33 17|
|9546291887|Edyth Deveau        |+(907) 101 70 64|
|9546291887|Jene Norgard        |+(694) 918 17 44|
|9168453994|Timika Bourke       |+(768) 691 44 85|
|9168453994|Lynell Riojas       |+(695) 918 33 16|
|4225784815|Jene Mackinnon      |+(540) 937 33 71|
+----------+--------------------+----------------+

val df = spark
  .read
  .format("cobol")
  .option("copybook_contents", copybook)
  .option("record_format", "V")
  .option("segment_field", "SEGMENT_ID")
  .option("segment_id_level0", "C")
  .option("segment_id_level1", "P")
  .load("examples/multisegment_data/COMP.DETAILS.SEP30.DATA.dat")

val df = spark
  .read
  .format("cobol")
  .option("copybook_contents", copybook)
  .option("record_format", "V")
  .option("segment_field", "SEGMENT_ID")
  .option("segment_id_level0", "C")
  .option("segment_id_level1", "_")
  .load("examples/multisegment_data/COMP.DETAILS.SEP30.DATA.dat")

df.show(10)
+------------------+-----------------------+----------+----------+--------------------+--------------------+
|           Seg_Id0|                Seg_Id1|SEGMENT_ID|COMPANY_ID|      STATIC_DETAILS|            CONTACTS|
+------------------+-----------------------+----------+----------+--------------------+--------------------+
|20181219130609_0_0|                   null|         C|9377942526|[Joan Q & Z,10 Sa...|[Joan Q & Z     1...|
|20181219130609_0_0|20181219130723_0_0_L1_1|         P|9377942526|[+(277) 944 44 5,...|[+(277) 944 44 55...|
|20181219130609_0_2|                   null|         C|3483483977|[Robotrd Inc.,2 P...|[Robotrd Inc.   2...|
|20181219130609_0_2|20181219130723_0_2_L1_1|         P|3483483977|[+(174) 970 97 5,...|[+(174) 970 97 54...|
|20181219130609_0_2|20181219130723_0_2_L1_2|         P|3483483977|[+(848) 832 61 6,...|[+(848) 832 61 68...|
|20181219130609_0_2|20181219130723_0_2_L1_3|         P|3483483977|[+(455) 184 13 3,...|[+(455) 184 13 39...|
|20181219130609_0_6|                   null|         C|7540764401|[Eqartion Inc.,87...|[Eqartion Inc.  8...|
|20181219130609_0_7|                   null|         C|4413124035|[Xingzhoug,74 Qin...|[Xingzhoug      7...|
|20181219130609_0_8|                   null|         C|9546291887|[ZjkLPj,5574, Tok...|[ZjkLPj         5...|
|20181219130609_0_8|20181219130723_0_8_L1_1|         P|9546291887|[+(300) 252 33 1,...|[+(300) 252 33 17...|
+------------------+-----------------------+----------+----------+--------------------+--------------------+