spark-sql-spark技术分享-第3页

HadoopFileLinesReader

2013-07-11admin阅读(1018)

HadoopFileLinesReader

HadoopFileLinesReader is a Scala Iterator of Apache Hadoop’s org.apache.hadoop.io.Text.

HadoopFileLinesReader is created to access datasets in the following data sources:

SimpleTextSource
LibSVMFileFormat
TextInputCSVDataSource
TextInputJsonDataSource
TextFileFormat

HadoopFileLinesReader uses the internal iterator that handles accessing files using Hadoop’s FileSystem API.

Creating HadoopFileLinesReader Instance

HadoopFileLinesReader takes the following when created:

PartitionedFile
Hadoop’s Configuration

`iterator` Internal Property



iterator: RecordReaderIterator[Text]

1

2

3

4

5

iterator: RecordReaderIterator[Text]

When created, HadoopFileLinesReader creates an internal iterator that uses Hadoop’s org.apache.hadoop.mapreduce.lib.input.FileSplit with Hadoop’s org.apache.hadoop.fs.Path and file.

iterator creates Hadoop’s TaskAttemptID, TaskAttemptContextImpl and LineRecordReader.

iterator initializes LineRecordReader and passes it on to RecordReaderIterator.

Note	`iterator` is used for `Iterator`-specific methods, i.e. `hasNext`, `next` and `close`.

OffHeapColumnVector

2013-07-10admin阅读(1381)

OffHeapColumnVector

OffHeapColumnVector is a concrete WritableColumnVector that…FIXME

Allocating Column Vectors — `allocateColumns` Static Method



OffHeapColumnVector[] allocateColumns(int capacity, StructType schema) (1)
OffHeapColumnVector[] allocateColumns(int capacity, StructField[] fields)

1

2

3

4

5

6

OffHeapColumnVector[] allocateColumns(int capacity, StructType schema) (1)

OffHeapColumnVector[] allocateColumns(int capacity, StructField[] fields)

Simply converts StructType to StructField[] and calls the other allocateColumns

allocateColumns creates an array of OffHeapColumnVector for every field (to hold capacity number of elements of the data type per field).

Note	`allocateColumns` is used when…FIXME

OnHeapColumnVector

2013-07-09admin阅读(1421)

OnHeapColumnVector

OnHeapColumnVector is a concrete WritableColumnVector that…FIXME

OnHeapColumnVector is created when:

OnHeapColumnVector is requested to allocate column vectors and reserveNewColumn
OrcColumnarBatchReader is requested to initBatch

Allocating Column Vectors — `allocateColumns` Static Method



OnHeapColumnVector[] allocateColumns(int capacity, StructType schema) (1)
OnHeapColumnVector[] allocateColumns(int capacity, StructField[] fields)

1

2

3

4

5

6

OnHeapColumnVector[] allocateColumns(int capacity, StructType schema) (1)

OnHeapColumnVector[] allocateColumns(int capacity, StructField[] fields)

Simply converts StructType to StructField[] and calls the other allocateColumns

allocateColumns creates an array of OnHeapColumnVector for every field (to hold capacity number of elements of the data type per field).

Note

allocateColumns is used when:

AggregateHashMap is created
InMemoryTableScanExec is requested to createAndDecompressColumn
VectorizedParquetRecordReader is requested to initBatch (with ON_HEAP memory mode)
OrcColumnarBatchReader is requested to initBatch (with ON_HEAP memory mode)
ColumnVectorUtils is requested to convert an iterator of rows into a single ColumnBatch (aka toBatch)

Creating OnHeapColumnVector Instance

OnHeapColumnVector takes the following when created:

Number of elements to hold in a vector (aka capacity)
Data type of the elements stored

When created, OnHeapColumnVector reserveInternal (for the given capacity) and reset.

`reserveInternal` Method



void reserveInternal(int newCapacity)

1

2

3

4

5

void reserveInternal(int newCapacity)

Note	`reserveInternal` is part of WritableColumnVector Contract to…FIXME.

reserveInternal…FIXME

`reserveNewColumn` Method



OnHeapColumnVector reserveNewColumn(int capacity, DataType type)

1

2

3

4

5

OnHeapColumnVector reserveNewColumn(int capacity, DataType type)

Note	`reserveNewColumn` is part of WritableColumnVector Contract to…FIXME.

reserveNewColumn…FIXME

WritableColumnVector

2013-07-08admin阅读(1404)

WritableColumnVector

WritableColumnVector is the contract for…FIXME



package org.apache.spark.sql.execution.vectorized;

abstract class WritableColumnVector extends ColumnVector {
  // only required methods that have no implementation
  // the others follow
  public abstract int getDictId(int rowId);
  protected abstract void reserveInternal(int capacity);
  public abstract void putNotNull(int rowId);
  public abstract void putNull(int rowId);
  public abstract void putNulls(int rowId, int count);
  public abstract void putNotNulls(int rowId, int count);
  // other type-specific puts
  protected abstract WritableColumnVector reserveNewColumn(int capacity, DataType type);
}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

package org.apache.spark.sql.execution.vectorized;

abstract class WritableColumnVector extends ColumnVector {

// only required methods that have no implementation

// the others follow

public abstract int getDictId(int rowId);

protected abstract void reserveInternal(int capacity);

public abstract void putNotNull(int rowId);

public abstract void putNull(int rowId);

public abstract void putNulls(int rowId, int count);

public abstract void putNotNulls(int rowId, int count);

// other type-specific puts

protected abstract WritableColumnVector reserveNewColumn(int capacity, DataType type);

}

Table 1. (Subset of) WritableColumnVector Contract
Method	Description
`reserveInternal`	Used when: OnHeapColumnVector and `OffHeapColumnVector` are created `WritableColumnVector` is requested to reserve
`reserveNewColumn`	Used when `WritableColumnVector` is created and requested to reserveDictionaryIds

Table 2. WritableColumnVectors
WritableColumnVector	Description
OffHeapColumnVector
OnHeapColumnVector

`reset` Method



void reset()

1

2

3

4

5

void reset()

reset…FIXME

Note	`reset` is used when…FIXME

`reserve` Method



void reserve(int requiredCapacity)

1

2

3

4

5

void reserve(int requiredCapacity)

reserve…FIXME

Note	`reserve` is used when…FIXME

`reserveDictionaryIds` Method



WritableColumnVector reserveDictionaryIds(int capacity)

1

2

3

4

5

WritableColumnVector reserveDictionaryIds(int capacity)

reserveDictionaryIds…FIXME

Note	`reserveDictionaryIds` is used when…FIXME

Creating WritableColumnVector Instance

WritableColumnVector takes the following when created:

Number of elements to hold in a vector (aka capacity)
Data type of the elements stored

WritableColumnVector initializes the internal registries and counters.

WritableColumnVector…FIXME

Note	`WritableColumnVector` is a Java abstract class and cannot be created directly.

ColumnVector

2013-07-07admin阅读(1542)

ColumnVector

ColumnVector is the contract for…FIXME



package org.apache.spark.sql.vectorized;

abstract class ColumnVector implements AutoCloseable {
  // only required methods that have no implementation
  // the others follow
  public abstract void close();
  public abstract boolean hasNull();
  public abstract int numNulls();
  public abstract boolean isNullAt(int rowId);
  public abstract boolean getBoolean(int rowId);
  // other type-specific getters
  protected abstract ColumnVector getChild(int ordinal);
}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

package org.apache.spark.sql.vectorized;

abstract class ColumnVector implements AutoCloseable {

// only required methods that have no implementation

// the others follow

public abstract void close();

public abstract boolean hasNull();

public abstract int numNulls();

public abstract boolean isNullAt(int rowId);

public abstract boolean getBoolean(int rowId);

// other type-specific getters

protected abstract ColumnVector getChild(int ordinal);

}

ColumnVector has a data type that you can access using dataType method.



DataType dataType()

1

2

3

4

5

DataType dataType()

Note	`ColumnVector` is an `Evolving` contract that is evolving towards becoming a stable API, but is not a stable API yet and can change from one feature release to another release. In other words, using the contract is as treading on thin ice.

Table 1. (Subset of) ColumnVector Contract
Method	Description
`getChild`	Used when…FIXME

Table 2. ColumnVectors
ColumnVector	Description
`ArrowColumnVector`
OffHeapColumnVector
OnHeapColumnVector
`OrcColumnVector`
WritableColumnVector

PartitioningUtils

2013-07-06admin阅读(1192)

PartitioningUtils Helper Object

PartitioningUtils is…FIXME

`validatePartitionColumn` Method



validatePartitionColumn(
  schema: StructType,
  partitionColumns: Seq[String],
  caseSensitive: Boolean): Unit

1

2

3

4

5

6

7

8

validatePartitionColumn(

schema: StructType,

partitionColumns: Seq[String],

caseSensitive: Boolean): Unit

validatePartitionColumn…FIXME

Note	`validatePartitionColumn` is used when…FIXME

`parsePartitions` Method



parsePartitions(
  paths: Seq[Path],
  typeInference: Boolean,
  basePaths: Set[Path],
  timeZoneId: String): PartitionSpec  (1)
parsePartitions(
  paths: Seq[Path],
  typeInference: Boolean,
  basePaths: Set[Path],
  timeZone: TimeZone): PartitionSpec

1

2

3

4

5

6

7

8

9

10

11

12

13

14

parsePartitions(

paths: Seq[Path],

typeInference: Boolean,

basePaths: Set[Path],

timeZoneId: String): PartitionSpec (1)

parsePartitions(

paths: Seq[Path],

typeInference: Boolean,

basePaths: Set[Path],

timeZone: TimeZone): PartitionSpec

Uses parsePartitions with timeZoneId mapped to a TimeZone

parsePartitions…FIXME

Note	`parsePartitions` is used when…FIXME

PartitionedFile — Part of Single File

2013-07-05admin阅读(1764)

PartitionedFile — Part of Single File

PartitionedFile is a part (aka block) of a single file that should be read, along with partition column values that need to be appended to each row.

Note	Partition column values are values of the columns that are column partitions and therefore part of the directory structure not the partitioned files themselves (that together are the partitioned dataset).

PartitionedFile is created exclusively when FileSourceScanExec is requested to create RDDs for bucketed or non-bucketed reads.

PartitionedFile uses the following text representation (i.e. toString):



path: [filePath], range: [start]-[end], partition values: [partitionValues]

1

2

3

4

5

path: [filePath], range: [start]-[end], partition values: [partitionValues]



import org.apache.spark.sql.execution.datasources.PartitionedFile
import org.apache.spark.sql.catalyst.InternalRow
val partFile = PartitionedFile(InternalRow.empty, "fakePath0", 0, 10, Array("host0", "host1"))

scala> println(partFile)
path: fakePath0, range: 0-10, partition values: [empty row]

1

2

3

4

5

6

7

8

9

10

import org.apache.spark.sql.execution.datasources.PartitionedFile

import org.apache.spark.sql.catalyst.InternalRow

val partFile = PartitionedFile(InternalRow.empty, "fakePath0", 0, 10, Array("host0", "host1"))

scala> println(partFile)

path: fakePath0, range: 0-10, partition values: [empty row]

Creating PartitionedFile Instance

PartitionedFile takes the following when created:

Partition column values to be appended to each row (as an internal row)
Path of the file to read
Beginning offset (in bytes)
Number of bytes to read (aka length)
Locality information as a list of nodes that have the data (aka locations). Empty by default.

IdentityBroadcastMode

2013-07-04admin阅读(1466)

IdentityBroadcastMode

IdentityBroadcastMode is a BroadcastMode that…FIXME

HashedRelationBroadcastMode

2013-07-03admin阅读(1697)

HashedRelationBroadcastMode

HashedRelationBroadcastMode is a BroadcastMode that BroadcastHashJoinExec uses for the required output distribution of child operators.

HashedRelationBroadcastMode takes build-side join keys (as Catalyst expressions) when created.

HashedRelationBroadcastMode gives a copy of itself with keys canonicalized when requested for a canonicalized version.

`transform` Method



transform(rows: Array[InternalRow]): HashedRelation (1)
transform(
  rows: Iterator[InternalRow],
  sizeHint: Option[Long]): HashedRelation

1

2

3

4

5

6

7

8

transform(rows: Array[InternalRow]): HashedRelation (1)

transform(

rows: Iterator[InternalRow],

sizeHint: Option[Long]): HashedRelation

Uses the other transform with the size of rows as sizeHint

Note	`transform` is part of BroadcastMode Contract to…FIXME.

transform…FIXME

BroadcastMode

2013-07-02admin阅读(1531)

BroadcastMode

BroadcastMode is the contract for…FIXME



package org.apache.spark.sql.catalyst.plans.physical

trait BroadcastMode {
  def canonicalized: BroadcastMode
  def transform(rows: Array[InternalRow]): Any
  def transform(rows: Iterator[InternalRow], sizeHint: Option[Long]): Any
}

1

2

3

4

5

6

7

8

9

10

11

package org.apache.spark.sql.catalyst.plans.physical

trait BroadcastMode {

def canonicalized: BroadcastMode

def transform(rows: Array[InternalRow]): Any

def transform(rows: Iterator[InternalRow], sizeHint: Option[Long]): Any

}

Method Description

canonicalized

Used when…FIXME

transform

Used when:

BroadcastExchangeExec is requested for relationFuture for the first time (when BroadcastExchangeExec is requested to prepare for execution as part of executing a physical operator)
HashedRelationBroadcastMode is requested to transform internal rows (and build a HashedRelation)

Note	The `rows`-only variant does not seem to be used at all.

Table 2. BroadcastModes
BroadcastMode	Description
HashedRelationBroadcastMode
IdentityBroadcastMode

spark-sql 第3页

HadoopFileLinesReader

HadoopFileLinesReader

Creating HadoopFileLinesReader Instance

`iterator` Internal Property

OffHeapColumnVector

OffHeapColumnVector

Allocating Column Vectors — `allocateColumns` Static Method

OnHeapColumnVector

OnHeapColumnVector

Allocating Column Vectors — `allocateColumns` Static Method

Creating OnHeapColumnVector Instance

`reserveInternal` Method

`reserveNewColumn` Method

WritableColumnVector

WritableColumnVector

`reset` Method

`reserve` Method

`reserveDictionaryIds` Method

Creating WritableColumnVector Instance

ColumnVector

ColumnVector

PartitioningUtils

PartitioningUtils Helper Object

`validatePartitionColumn` Method

`parsePartitions` Method

PartitionedFile — Part of Single File

PartitionedFile — Part of Single File

Creating PartitionedFile Instance

IdentityBroadcastMode

IdentityBroadcastMode

HashedRelationBroadcastMode

HashedRelationBroadcastMode

`transform` Method

BroadcastMode

BroadcastMode

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

spark-sql 第3页

HadoopFileLinesReader

Creating HadoopFileLinesReader Instance

iterator Internal Property

OffHeapColumnVector

Allocating Column Vectors — allocateColumns Static Method

OnHeapColumnVector

Allocating Column Vectors — allocateColumns Static Method

Creating OnHeapColumnVector Instance

reserveInternal Method

reserveNewColumn Method

WritableColumnVector

reset Method

reserve Method

reserveDictionaryIds Method

Creating WritableColumnVector Instance

ColumnVector

PartitioningUtils Helper Object

validatePartitionColumn Method

parsePartitions Method

PartitionedFile — Part of Single File

Creating PartitionedFile Instance

IdentityBroadcastMode

HashedRelationBroadcastMode

transform Method

BroadcastMode

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

`iterator` Internal Property

Allocating Column Vectors — `allocateColumns` Static Method

Allocating Column Vectors — `allocateColumns` Static Method

`reserveInternal` Method

`reserveNewColumn` Method

`reset` Method

`reserve` Method

`reserveDictionaryIds` Method

`validatePartitionColumn` Method

`parsePartitions` Method

`transform` Method