<commit_before># Skale Reference

<!-- START doctoc generated TOC please keep comment here to allow auto update -->
<!-- DON'T EDIT THIS SECTION, INSTEAD RE-RUN doctoc TO UPDATE -->


- [Overview](#overview)
- [Working with datasets](#working-with-datasets)
- [Skale module](#skale-module)
  - [skale.context([config])](#skalecontextconfig)
    - [sc.end()](#scend)
    - [sc.parallelize(array)](#scparallelizearray)
    - [sc.range(start[, end[, step]])](#scrangestart-end-step)
    - [sc.textFile(path)](#sctextfilepath)
    - [sc.lineStream(input_stream)](#sclinestreaminput_stream)
    - [sc.objectStream(input_stream)](#scobjectstreaminput_stream)
  - [Dataset methods](#dataset-methods)
    - [ds.aggregate(reducer, combiner, init[,obj])](#dsaggregatereducer-combiner-initobj)
    - [ds.aggregateByKey(reducer, combiner, init,[ obj])](#dsaggregatebykeyreducer-combiner-init-obj)
    - [ds.cartesian(other)](#dscartesianother)
    - [ds.coGroup(other)](#dscogroupother)
    - [ds.collect([opt])](#dscollectopt)
    - [ds.count()](#dscount)
    - [ds.countByKey()](#dscountbykey)
    - [ds.countByValue()](#dscountbyvalue)
    - [ds.distinct()](#dsdistinct)
    - [ds.filter(filter[,obj])](#dsfilterfilterobj)
    - [ds.first()](#dsfirst)
    - [ds.flatMap(flatMapper[,obj])](#dsflatmapflatmapperobj)
    - [ds.flatMapValues(flatMapper[,obj])](#dsflatmapvaluesflatmapperobj)
    - [ds.foreach(callback[, obj])](#dsforeachcallback-obj)
    - [ds.groupByKey()](#dsgroupbykey)
    - [ds.intersection(other)](#dsintersectionother)
    - [ds.join(other)](#dsjoinother)
    - [ds.keys()](#dskeys)
    - [ds.leftOuterJoin(other)](#dsleftouterjoinother)
    - [ds.lookup(k)](#dslookupk)
    - [ds.map(mapper[,obj])](#dsmapmapperobj)
    - [ds.mapValues(mapper[,obj])](#dsmapvaluesmapperobj)
    - [ds.partitionBy(partitioner)](#dspartitionbypartitioner)
    - [ds.persist()](#dspersist)
    - [ds.reduce(reducer, init[,obj])](#dsreducereducer-initobj)
    - [ds.reduceByKey(reducer, init[, obj])](#dsreducebykeyreducer-init-obj)
    - [ds.rightOuterJoin(other)](#dsrightouterjoinother)
    - [ds.sample(withReplacement, frac, seed)](#dssamplewithreplacement-frac-seed)
    - [ds.sortBy(keyfunc[, ascending])](#dssortbykeyfunc-ascending)
    - [ds.sortByKey(ascending)](#dssortbykeyascending)
    - [ds.subtract(other)](#dssubtractother)
    - [ds.take(num)](#dstakenum)
    - [ds.top(num)](#dstopnum)
    - [ds.union(other)](#dsunionother)
    - [ds.values()](#dsvalues)
- [References](#references)

<!-- END doctoc generated TOC please keep comment here to allow auto update -->

## Overview

Skale is a fast and general purpose distributed data processing
system. It provides a high-level API in Javascript and an optimized
parallel execution engine.

A Skale application consists of a *master* program that runs the
user code and executes various *parallel operations* on a cluster
of *workers*.

The main abstraction Skale provides is a *dataset* which is similar
to a Javascript *array*, but partitioned accross the workers that
can be operated in parallel.

There are several ways to create a dataset: *parallelizing* an existing
array in the master program, or referencing a dataset in a distributed
storage system (such as HDFS), or *streaming* the content of any
source that can be processed through Node.js *Streams*. We call
*source* a function which initializes a dataset.

Datasets support two kinds of operations: *transformations*, which create
a new dataset from an existing one, and *actions*, which
return a value to the *master* program after running a computation
on the dataset.

For example, `map` is a transformation that applies a function to
each element of a dataset, returning a new dataset. On the other
hand, `reduce` is an action that aggregates all elements of a dataset
using some function, and returns the final result to the master.

*Sources* and *transformations* in Skale are *lazy*. They do not
start right away, but are triggered by *actions*, thus allowing
efficient pipelined execution and optimized data transfers.

A first example:

```javascript
var sc = require('skale-engine').context();		// create a new context
sc.parallelize([1, 2, 3, 4]).				// source
   map(function (x) {return x+1}).			// transform
   reduce(function (a, b) {return a+b}, 0).	// action
   on('data', console.log);					// process result: 14
```

## Working with datasets
<a name=working-with-datasets></a>

After having initialized a cluster context using
[skale.context()](#skale-context), one can create a dataset
using the following sources:

| Source Name                                       | Description                                           |
| ------------------------------------------------- | ----------------------------------------------------- |
|[lineStream(stream)](#sclinestreaminput_stream)    | Create a dataset from a text stream                   |
|[objectStream(stream)](#scobjectstreaminput_stream)| Create a dataset from an object stream                |
|[parallelize(array)](#scparallelizearray)          | Create a dataset from an array                        |
|[range(start,end,step)](#scrangestart-end-step)    | Create a dataset containing integers from start to end|
|[textFile(path)](#sctextfilepath)                  | Create a dataset from a regular text file             |

Transformations operate on a dataset and return a new dataset. Note that some
transformation operate only on datasets where each element is in the form
of 2 elements array of key and value (`[k,v]` dataset):

	[[Ki,Vi], ..., [Kj, Vj]]

A special transformation `persist()` enables one to *persist* a dataset
in memory, allowing efficient reuse accross parallel operations.

|Transformation Name              | Description                                   | in    | out   |
| -----------------               |-----------------------------------------------|-------|-------|
|[cartesian(other)](#dscartesianother) | Perform a cartesian product with the other dataset | v w | [v,w]|
|[coGroup(other)](#dscogroupother) | Group data from both datasets sharing the same key | [k,v] [k,w] |[k,[[v],[w]]]|
|[distinct()](#dsdistinct)    | Return a dataset where duplicates are removed | v | w|
|[filter(func)](#dsfilterfilterobj)| Return a dataset of elements on which function returns true | v | w|
|[flatMap(func)](#dsflatmapflatmapperobj)| Pass the dataset elements to a function which returns a sequence | v | w|
|[groupByKey()](#dsgroupbykey)| Group values with the same key | [k,v] | [k,[v]]|
|[intersection(other)](#dsintersection) | Return a dataset containing only elements found in both datasets | v w | v|
|[join(other)](#dsjoinother)       | Perform an inner join between 2 datasets | [k,v] | [k,[v,w]]|
|[leftOuterJoin(other)](#dsleftouterjoinother) | Join 2 datasets where the key must be present in the other | [k,v] | [k,[v,w]]|
|[rightOuterJoin(other)](#dsrightouterjoinother) | Join 2 datasets where the key must be present in the first | [k,v] | [k,[v,w]]|
|[keys()](#dskeys)            | Return a dataset of just the keys | [k,v] | k|
|[map(func)](#dsmapmapperobj) | Return a dataset where elements are passed through a function | v | w|
|[mapValues(func)](#dsmapvaluesmapperobj)| Map a function to the value field of key-value dataset | [k,v] | [k,w]|
|[reduceByKey(func, init)](#dsreducebykeyreducer-init-obj)| Combine values with the same key | [k,v] | [k,w]|
|[partitionBy(partitioner)](#dspartitionbypartitioner)| Partition using the partitioner | v | v|
|[persist()](#dspersist)      | Idempotent. Keep content of dataset in cache for further reuse. | v | v|
|[sample(rep, frac, seed)](#dssamplewithreplacement-frac-seed) | Sample a dataset, with or without replacement | v | w|
|[sortBy(func)](#dssortbykeyfunc-ascending) | Sort a dataset | v | v|
|[sortByKey()](#dssortbykeyascending) | Sort a [k,v] dataset | [k,v] | [k,v]|
|[subtract(other)](#dssubractother) | Remove the content of one dataset | v w | v|
|[union(other)](#dsunionother)     | Return a dataset containing elements from both datasets | v | v w|
|[values()](#dsvalues)        | Return a dataset of just the values | [k,v] | v|

Actions operate on a dataset and send back results to the *master*. Results
are always produced asynchronously. All actions return a [readable stream]
on which results are emitted.

|Action Name | Description | out|
|------------------             |----------------------------------------------|--------------|
|[aggregate(func, func, init)](#dsaggregatereducer-combiner-initobj)| Similar to reduce() but may return a different type| stream of value |
|[aggregateByKey(func, func, init)](#dsaggregatebykeyreducer-combiner-init-obj)| reduce and combine by key using functions| stream of [k,v] |
|[collect()](#dscollectopt)         | Return the content of dataset | stream of elements|
|[count()](#dscount)             | Return the number of elements from dataset | stream of number|
|[countByKey()](#dscountbykey)     | Return the number of occurrences for each key in a `[k,v]` dataset | stream of [k,number]|
|[countByValue()](#dscountbyvalue) | Return the number of occurrences of elements from dataset | stream of [v,number]|
|[first()](#first)               | Return the first element in dataset | stream of value |
|[foreach(func)](#dsforeachcallback-obj)| Apply the provided function to each element of the dataset | empty stream |
|[lookup(k)](#dslookupk)          | Return the list of values `v` for key `k` in a `[k,v]` dataset | stream of v|
|[reduce(func, init)](#dsreducereducer-initobj)| Aggregates dataset elements using a function into one value | stream of value|
|[take(num)](#dstakenum)         | Return the first `num` elements of dataset | stream of value|
|[top(num)](#dstopnum)           | Return the top `num` elements of dataset | stream of value|

## Skale module

The Skale module is the main entry point for Skale functionality.
To use it, one must `require('skale-engine')`.

### skale.context([config])

Creates and returns a new context which represents the connection
to the Skale cluster, and which can be used to create datasets on that
cluster. Config is an *Object* which defines the cluster server,
with the following defaults:

```javascript
{
  host: 'localhost',	// Cluster server host, settable also by SKALE_HOST env
  port: '12346'			// Cluster server port, settable also by SKALE_PORT env
}
```<commit_msg>doc: fix cross reference links<commit_after># Skale Reference

<!-- START doctoc generated TOC please keep comment here to allow auto update -->
<!-- DON'T EDIT THIS SECTION, INSTEAD RE-RUN doctoc TO UPDATE -->


- [Overview](#overview)
- [Working with datasets](#working-with-datasets)
- [Skale module](#skale-module)
  - [skale.context([config])](#skalecontextconfig)
    - [sc.end()](#scend)
    - [sc.parallelize(array)](#scparallelizearray)
    - [sc.range(start[, end[, step]])](#scrangestart-end-step)
    - [sc.textFile(path)](#sctextfilepath)
    - [sc.lineStream(input_stream)](#sclinestreaminput_stream)
    - [sc.objectStream(input_stream)](#scobjectstreaminput_stream)
  - [Dataset methods](#dataset-methods)
    - [ds.aggregate(reducer, combiner, init[,obj])](#dsaggregatereducer-combiner-initobj)
    - [ds.aggregateByKey(reducer, combiner, init,[ obj])](#dsaggregatebykeyreducer-combiner-init-obj)
    - [ds.cartesian(other)](#dscartesianother)
    - [ds.coGroup(other)](#dscogroupother)
    - [ds.collect([opt])](#dscollectopt)
    - [ds.count()](#dscount)
    - [ds.countByKey()](#dscountbykey)
    - [ds.countByValue()](#dscountbyvalue)
    - [ds.distinct()](#dsdistinct)
    - [ds.filter(filter[,obj])](#dsfilterfilterobj)
    - [ds.first()](#dsfirst)
    - [ds.flatMap(flatMapper[,obj])](#dsflatmapflatmapperobj)
    - [ds.flatMapValues(flatMapper[,obj])](#dsflatmapvaluesflatmapperobj)
    - [ds.foreach(callback[, obj])](#dsforeachcallback-obj)
    - [ds.groupByKey()](#dsgroupbykey)
    - [ds.intersection(other)](#dsintersectionother)
    - [ds.join(other)](#dsjoinother)
    - [ds.keys()](#dskeys)
    - [ds.leftOuterJoin(other)](#dsleftouterjoinother)
    - [ds.lookup(k)](#dslookupk)
    - [ds.map(mapper[,obj])](#dsmapmapperobj)
    - [ds.mapValues(mapper[,obj])](#dsmapvaluesmapperobj)
    - [ds.partitionBy(partitioner)](#dspartitionbypartitioner)
    - [ds.persist()](#dspersist)
    - [ds.reduce(reducer, init[,obj])](#dsreducereducer-initobj)
    - [ds.reduceByKey(reducer, init[, obj])](#dsreducebykeyreducer-init-obj)
    - [ds.rightOuterJoin(other)](#dsrightouterjoinother)
    - [ds.sample(withReplacement, frac, seed)](#dssamplewithreplacement-frac-seed)
    - [ds.sortBy(keyfunc[, ascending])](#dssortbykeyfunc-ascending)
    - [ds.sortByKey(ascending)](#dssortbykeyascending)
    - [ds.subtract(other)](#dssubtractother)
    - [ds.take(num)](#dstakenum)
    - [ds.top(num)](#dstopnum)
    - [ds.union(other)](#dsunionother)
    - [ds.values()](#dsvalues)
- [References](#references)

<!-- END doctoc generated TOC please keep comment here to allow auto update -->

## Overview

Skale is a fast and general purpose distributed data processing
system. It provides a high-level API in Javascript and an optimized
parallel execution engine.

A Skale application consists of a *master* program that runs the
user code and executes various *parallel operations* on a cluster
of *workers*.

The main abstraction Skale provides is a *dataset* which is similar
to a Javascript *array*, but partitioned accross the workers that
can be operated in parallel.

There are several ways to create a dataset: *parallelizing* an existing
array in the master program, or referencing a dataset in a distributed
storage system (such as HDFS), or *streaming* the content of any
source that can be processed through Node.js *Streams*. We call
*source* a function which initializes a dataset.

Datasets support two kinds of operations: *transformations*, which create
a new dataset from an existing one, and *actions*, which
return a value to the *master* program after running a computation
on the dataset.

For example, `map` is a transformation that applies a function to
each element of a dataset, returning a new dataset. On the other
hand, `reduce` is an action that aggregates all elements of a dataset
using some function, and returns the final result to the master.

*Sources* and *transformations* in Skale are *lazy*. They do not
start right away, but are triggered by *actions*, thus allowing
efficient pipelined execution and optimized data transfers.

A first example:

```javascript
var sc = require('skale-engine').context();		// create a new context
sc.parallelize([1, 2, 3, 4]).				// source
   map(function (x) {return x+1}).			// transform
   reduce(function (a, b) {return a+b}, 0).	// action
   on('data', console.log);					// process result: 14
```

## Working with datasets
<a name=working-with-datasets></a>

After having initialized a cluster context using
[skale.context()](#skale-context), one can create a dataset
using the following sources:

| Source Name                                       | Description                                           |
| ------------------------------------------------- | ----------------------------------------------------- |
|[lineStream(stream)](#sclinestreaminput_stream)    | Create a dataset from a text stream                   |
|[objectStream(stream)](#scobjectstreaminput_stream)| Create a dataset from an object stream                |
|[parallelize(array)](#scparallelizearray)          | Create a dataset from an array                        |
|[range(start,end,step)](#scrangestart-end-step)    | Create a dataset containing integers from start to end|
|[textFile(path)](#sctextfilepath)                  | Create a dataset from a regular text file             |

Transformations operate on a dataset and return a new dataset. Note that some
transformation operate only on datasets where each element is in the form
of 2 elements array of key and value (`[k,v]` dataset):

	[[Ki,Vi], ..., [Kj, Vj]]

A special transformation `persist()` enables one to *persist* a dataset
in memory, allowing efficient reuse accross parallel operations.

|Transformation Name              | Description                                   | in    | out   |
| -----------------               |-----------------------------------------------|-------|-------|
|[cartesian(other)](#dscartesianother) | Perform a cartesian product with the other dataset | v w | [v,w]|
|[coGroup(other)](#dscogroupother) | Group data from both datasets sharing the same key | [k,v] [k,w] |[k,[[v],[w]]]|
|[distinct()](#dsdistinct)    | Return a dataset where duplicates are removed | v | w|
|[filter(func)](#dsfilterfilterobj)| Return a dataset of elements on which function returns true | v | w|
|[flatMap(func)](#dsflatmapflatmapperobj)| Pass the dataset elements to a function which returns a sequence | v | w|
|[groupByKey()](#dsgroupbykey)| Group values with the same key | [k,v] | [k,[v]]|
|[intersection(other)](#dsintersectionother) | Return a dataset containing only elements found in both datasets | v w | v|
|[join(other)](#dsjoinother)       | Perform an inner join between 2 datasets | [k,v] | [k,[v,w]]|
|[leftOuterJoin(other)](#dsleftouterjoinother) | Join 2 datasets where the key must be present in the other | [k,v] | [k,[v,w]]|
|[rightOuterJoin(other)](#dsrightouterjoinother) | Join 2 datasets where the key must be present in the first | [k,v] | [k,[v,w]]|
|[keys()](#dskeys)            | Return a dataset of just the keys | [k,v] | k|
|[map(func)](#dsmapmapperobj) | Return a dataset where elements are passed through a function | v | w|
|[mapValues(func)](#dsmapvaluesmapperobj)| Map a function to the value field of key-value dataset | [k,v] | [k,w]|
|[reduceByKey(func, init)](#dsreducebykeyreducer-init-obj)| Combine values with the same key | [k,v] | [k,w]|
|[partitionBy(partitioner)](#dspartitionbypartitioner)| Partition using the partitioner | v | v|
|[persist()](#dspersist)      | Idempotent. Keep content of dataset in cache for further reuse. | v | v|
|[sample(rep, frac, seed)](#dssamplewithreplacement-frac-seed) | Sample a dataset, with or without replacement | v | w|
|[sortBy(func)](#dssortbykeyfunc-ascending) | Sort a dataset | v | v|
|[sortByKey()](#dssortbykeyascending) | Sort a [k,v] dataset | [k,v] | [k,v]|
|[subtract(other)](#dssubtractother) | Remove the content of one dataset | v w | v|
|[union(other)](#dsunionother)     | Return a dataset containing elements from both datasets | v | v w|
|[values()](#dsvalues)        | Return a dataset of just the values | [k,v] | v|

Actions operate on a dataset and send back results to the *master*. Results
are always produced asynchronously. All actions return a [readable stream]
on which results are emitted.

|Action Name | Description | out|
|------------------             |----------------------------------------------|--------------|
|[aggregate(func, func, init)](#dsaggregatereducer-combiner-initobj)| Similar to reduce() but may return a different type| stream of value |
|[aggregateByKey(func, func, init)](#dsaggregatebykeyreducer-combiner-init-obj)| reduce and combine by key using functions| stream of [k,v] |
|[collect()](#dscollectopt)         | Return the content of dataset | stream of elements|
|[count()](#dscount)             | Return the number of elements from dataset | stream of number|
|[countByKey()](#dscountbykey)     | Return the number of occurrences for each key in a `[k,v]` dataset | stream of [k,number]|
|[countByValue()](#dscountbyvalue) | Return the number of occurrences of elements from dataset | stream of [v,number]|
|[first()](#dsfirst)               | Return the first element in dataset | stream of value |
|[foreach(func)](#dsforeachcallback-obj)| Apply the provided function to each element of the dataset | empty stream |
|[lookup(k)](#dslookupk)          | Return the list of values `v` for key `k` in a `[k,v]` dataset | stream of v|
|[reduce(func, init)](#dsreducereducer-initobj)| Aggregates dataset elements using a function into one value | stream of value|
|[take(num)](#dstakenum)         | Return the first `num` elements of dataset | stream of value|
|[top(num)](#dstopnum)           | Return the top `num` elements of dataset | stream of value|

## Skale module

The Skale module is the main entry point for Skale functionality.
To use it, one must `require('skale-engine')`.

### skale.context([config])

Creates and returns a new context which represents the connection
to the Skale cluster, and which can be used to create datasets on that
cluster. Config is an *Object* which defines the cluster server,
with the following defaults:

```javascript
{
  host: 'localhost',	// Cluster server host, settable also by SKALE_HOST env
  port: '12346'			// Cluster server port, settable also by SKALE_PORT env
}
```
{'fraction_non_alphanumeric': 0.15424893116203103, 'fraction_numerical': 0.001633280491905654, 'mean_word_length': 3.704632768361582, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2921392', 'n_tokens_mistral': 6532, 'n_tokens_neox': 6033, 'n_words': 2014}