Skip to content

data

KlinkerBlockManager

Class for handling of blocks.

Parameters:

Name Type Description Default
blocks DataFrame

dataframe with blocks.

required

Examples:

>>> from klinker import KlinkerBlockManager
>>> kbm = KlinkerBlockManager.from_dict({ "block1": [[1,3,4],[3,4,5]], "block2": [[3,4,5],[5,6]]}, dataset_names=("A","B"))
>>> kbm.blocks.compute()
                A          B
block1  [1, 3, 4]  [3, 4, 5]
block2  [3, 4, 5]     [5, 6]
>>> kbm["block1"].compute()
                A          B
block1  [1, 3, 4]  [3, 4, 5]
>>> len(kbm)
2
>>> set(kbm.all_pairs())
{(4, 4), (5, 5), (3, 4), (1, 5), (4, 3), (4, 6), (1, 4), (4, 5), (3, 3), (5, 6), (3, 6), (1, 3), (3, 5)}
>>> kbm.block_sizes
block1    6
block2    5
Name: block_sizes, dtype: int64
>>> kbm.mean_block_size
5.5
>>> kbm.to_dict()
{'block1': ([1, 3, 4], [3, 4, 5]), 'block2': ([3, 4, 5], [5, 6])}

```
Source code in klinker/data/blocks.py
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
class KlinkerBlockManager:
    """Class for handling of blocks.

    Args:
        blocks: dataframe with blocks.

    Examples:

        >>> from klinker import KlinkerBlockManager
        >>> kbm = KlinkerBlockManager.from_dict({ "block1": [[1,3,4],[3,4,5]], "block2": [[3,4,5],[5,6]]}, dataset_names=("A","B"))
        >>> kbm.blocks.compute()
                        A          B
        block1  [1, 3, 4]  [3, 4, 5]
        block2  [3, 4, 5]     [5, 6]
        >>> kbm["block1"].compute()
                        A          B
        block1  [1, 3, 4]  [3, 4, 5]
        >>> len(kbm)
        2
        >>> set(kbm.all_pairs())
        {(4, 4), (5, 5), (3, 4), (1, 5), (4, 3), (4, 6), (1, 4), (4, 5), (3, 3), (5, 6), (3, 6), (1, 3), (3, 5)}
        >>> kbm.block_sizes
        block1    6
        block2    5
        Name: block_sizes, dtype: int64
        >>> kbm.mean_block_size
        5.5
        >>> kbm.to_dict()
        {'block1': ([1, 3, 4], [3, 4, 5]), 'block2': ([3, 4, 5], [5, 6])}

        ```
    """

    def __init__(self, blocks: dd.DataFrame):
        self.blocks = blocks
        grouped = []
        for column_name in self.blocks.columns:
            cur_ex = self.blocks[column_name].explode()
            grouped.append(cur_ex.to_frame().groupby(by=column_name))
        self._grouped = tuple(grouped)

    def __getitem__(self, key):
        return self.blocks.loc[key]

    def __len__(self) -> int:
        return len(self.blocks)

    def __repr__(self) -> str:
        return f"KlinkerBlockManager(blocks=\n{self.blocks.__repr__()})"

    def to_dict(self) -> Dict[Union[str, int], Tuple[Union[str, int], Union[str, int]]]:
        """Return blocks as dict.

        Returns:
          The dict has block names as keys and a tuple of sets of entity ids.
        """
        return (
            self.blocks.apply(tuple, axis=1, meta=pd.Series([], dtype=object))
            .compute()
            .to_dict()
        )

    def find_blocks(self, entity_id: Union[str, int], column_id: int) -> np.ndarray:
        """Find blocks where entity id belongs to.

        Args:
          entity_id: Union[str, int]: Entity id.
          column_id: int: Whether entity belongs to left (0) or right (1) dataset.

        Returns:
            Blocks where entity id belongs to.
        """
        return self._grouped[column_id].get_group(entity_id).index.values.compute()

    def entity_pairs(
        self, entity_id: Union[str, int], column_id: int
    ) -> Generator[Tuple[Union[int, str], ...], None, None]:
        """Get all pairs where this entity shows up.

        Args:
          entity_id: Union[str, int]: Entity id.
          column_id: int: Whether entity belongs to left (0) or right (1) dataset.

        Returns:
            Generator for these pairs.
        """
        cur_blocks = self.find_blocks(entity_id, column_id)
        other_column = 0 if column_id == 1 else 1
        other_column_name = self.blocks.columns[other_column]
        return (
            pair
            for blk_name in cur_blocks
            for _, blk in self.blocks.loc[blk_name][other_column_name].compute().items()
            for pair in itertools.product({entity_id}, blk)
        )

    def all_pairs(self) -> Generator[Tuple[Union[int, str], ...], None, None]:
        """Get all pairs

        Returns:
            Generator that creates all pairs, from blocks (including duplicates).
        """
        for block_tuple in self.blocks.itertuples(index=False, name=None):
            for pair in itertools.product(*block_tuple):
                yield pair

    @property
    def block_sizes(self) -> pd.DataFrame:
        """Sizes of blocks"""
        meta = pd.Series([], dtype="int64", name="block_sizes")
        return self.blocks.apply(
            lambda x: sum(len(v) for v in x), axis=1, meta=meta
        ).compute()

    @property
    def mean_block_size(self) -> float:
        """Mean size of all blocks."""
        return self.block_sizes.mean()

    @classmethod
    def combine(
        cls, this: "KlinkerBlockManager", other: "KlinkerBlockManager"
    ) -> "KlinkerBlockManager":
        """Combine blocks.

        Args:
          this: one block manager to combine
          other: other block manager to combine

        Returns:
          Combined KlinkerBlockManager

        Examples:

            >>> from klinker import KlinkerBlockManager
            >>> kbm = KlinkerBlockManager.from_dict({"block1": [[1,3,4],[3,4,5]], "block2": [[3,4,5],[5,6]]}, dataset_names=("A","B"))
            >>> kbm2 = KlinkerBlockManager.from_dict({"block3": [[7,4],[12,8]]}, dataset_names=("A","B"))
            >>> kbm_merged = KlinkerBlockManager.combine(kbm, kbm2)
            >>> kbm_merged.blocks.compute()
                            A          B
            block1  [1, 3, 4]  [3, 4, 5]
            block2  [3, 4, 5]     [5, 6]
            block3     [7, 4]    [12, 8]

        """

        def _merge_blocks(
            row: pd.Series, output_names: Sequence[str], left_right_names: Sequence[str]
        ):
            nonnull = row[~row.isnull()]
            if len(nonnull) == 2:  # no block overlap
                nonnull.index = output_names
                return nonnull
            else:
                A_left = set(nonnull[left_right_names[0]])
                A_right = set(nonnull[left_right_names[2]])
                B_left = set(nonnull[left_right_names[1]])
                B_right = set(nonnull[left_right_names[3]])
                A = list(A_left.union(A_right))
                B = list(B_left.union(B_right))
                return pd.Series([A, B], index=output_names, name=nonnull.name)

        if list(this.blocks.columns) != list(other.blocks.columns):
            raise ValueError("Cannot combine blocks from different datasets!")

        output_names = this.blocks.columns
        left_suffix = "left"
        right_suffix = "right"
        left_right_names = [
            col + suffix
            for col_names, suffix in zip(
                [this.blocks.columns, other.blocks.columns], [left_suffix, right_suffix]
            )
            for col in col_names
        ]
        joined = this.blocks.join(
            other.blocks, how="outer", lsuffix="left", rsuffix="right"
        )

        meta = pd.DataFrame([], columns=output_names)
        return cls(
            joined.apply(
                _merge_blocks,
                output_names=output_names,
                left_right_names=left_right_names,
                axis=1,
                meta=meta,
            )
        )

    def to_parquet(self, path: Union[str, pathlib.Path], **kwargs):
        """Write blocks as parquet file(s).

        Args:
          path: Union[str, pathlib.Path]: Where to write.
          **kwargs: passed to the parquet function
        """
        if "schema" not in kwargs:
            left, right = self.blocks.columns[:2]
            block_type = pa.list_(pa.string())
            schema = {
                left: block_type,
                right: block_type,
            }
        else:
            schema = kwargs.pop["schema"]  # type: ignore
        try:
            self.blocks.to_parquet(path, schema=schema, **kwargs)
        except ValueError:
            # If index is incorrectly assumed by dask to be string
            # and it turns out to be int64 an error would be thrown
            # This is kind of a dirty hack
            schema["__null_dask_index__"] = pa.int64()
            self.blocks.to_parquet(path, schema=schema, **kwargs)

    @classmethod
    def read_parquet(
        cls,
        path: Union[str, pathlib.Path],
        calculate_divisions: bool = True,
        **kwargs,
    ) -> "KlinkerBlockManager":
        """Read blocks from parquet.

        Args:
          path: Union[str, pathlib.Path]: Path where blocks are stored.
          calculate_divisions: bool: Calculate index divisions.
          **kwargs: Passed to `dd.read_parquet` function.

        Returns:
            Blocks as KlinkerBlockManager
        """
        return cls(
            dd.read_parquet(
                path=path,
                calculate_divisions=calculate_divisions,
                **kwargs,
            )
        )

    @classmethod
    def from_pandas(
        cls, df: pd.DataFrame, npartitions: int = 1, **kwargs
    ) -> "KlinkerBlockManager":
        """Create from pandas.

        Args:
          df: pd.DataFrame: DataFrame
          npartitions: int:  Partitions for dask
          **kwargs: Passed to `dd.from_pandas`

        Returns:
            Blocks as KlinkerBlockManager

        Examples:

            >>> import pandas as pd
            >>> from klinker import KlinkerBlockManager
            >>> pd_blocks = pd.DataFrame({'A': {'block1': [1, 3, 4], 'block2': [3, 4, 5]}, 'B': {'block1': [3, 4, 5], 'block2': [5, 6]}})
            >>> kbm = KlinkerBlockManager.from_pandas(pd_blocks)

        """
        return cls(dd.from_pandas(df, npartitions=npartitions, **kwargs))

    @classmethod
    def from_dict(
        cls,
        block_dict: Dict[
            BlockIdTypeVar, Tuple[List[EntityIdTypeVar], List[EntityIdTypeVar]]
        ],
        dataset_names: Tuple[str, str] = ("left", "right"),
        npartitions: int = 1,
        **kwargs,
    ) -> "KlinkerBlockManager":
        """

        Args:
          block_dict: Dictionary with block information.
          dataset_names: Tuple[str, str]: Tuple of dataset names.
          npartitions: int: Partitions used for dask.
          **kwargs: Passed to `dd.from_dict`.

        Returns:
            Blocks as KlinkerBlockManager

        Examples:

            >>> from klinker import KlinkerBlockManager
            >>> kbm = KlinkerBlockManager.from_dict({"block1": [[1,3,4],[3,4,5]], "block2": [[3,4,5],[5,6]]}, dataset_names=("A","B"))

        """
        return cls(
            dd.from_dict(
                block_dict,
                orient="index",
                columns=dataset_names,
                npartitions=npartitions,
                **kwargs,
            )
        )

    @classmethod
    @deprecated(reason="Please use parquet files")
    def read_pickle(cls, path) -> "KlinkerBlockManager":
        with open(path, "rb") as in_file:
            res = pickle.load(in_file)
            if isinstance(res, dict):
                return cls.from_dict(res)
            elif isinstance(res, pd.DataFrame):
                return cls.from_pandas(res)
            elif hasattr(res, "blocks") and isinstance(res.blocks, dict):
                return cls.from_dict(
                    {
                        bk: (list(left_v), list(right_v))
                        for bk, (left_v, right_v) in res.blocks.items()
                    }
                )  # type: ignore
            else:
                raise ValueError(f"Unknown pickled object of type {type(res)}")

block_sizes: pd.DataFrame property

Sizes of blocks

mean_block_size: float property

Mean size of all blocks.

all_pairs()

Get all pairs

Returns:

Type Description
Generator[Tuple[Union[int, str], ...], None, None]

Generator that creates all pairs, from blocks (including duplicates).

Source code in klinker/data/blocks.py
422
423
424
425
426
427
428
429
430
def all_pairs(self) -> Generator[Tuple[Union[int, str], ...], None, None]:
    """Get all pairs

    Returns:
        Generator that creates all pairs, from blocks (including duplicates).
    """
    for block_tuple in self.blocks.itertuples(index=False, name=None):
        for pair in itertools.product(*block_tuple):
            yield pair

combine(this, other) classmethod

Combine blocks.

Parameters:

Name Type Description Default
this KlinkerBlockManager

one block manager to combine

required
other KlinkerBlockManager

other block manager to combine

required

Returns:

Type Description
KlinkerBlockManager

Combined KlinkerBlockManager

Examples:

>>> from klinker import KlinkerBlockManager
>>> kbm = KlinkerBlockManager.from_dict({"block1": [[1,3,4],[3,4,5]], "block2": [[3,4,5],[5,6]]}, dataset_names=("A","B"))
>>> kbm2 = KlinkerBlockManager.from_dict({"block3": [[7,4],[12,8]]}, dataset_names=("A","B"))
>>> kbm_merged = KlinkerBlockManager.combine(kbm, kbm2)
>>> kbm_merged.blocks.compute()
                A          B
block1  [1, 3, 4]  [3, 4, 5]
block2  [3, 4, 5]     [5, 6]
block3     [7, 4]    [12, 8]
Source code in klinker/data/blocks.py
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
@classmethod
def combine(
    cls, this: "KlinkerBlockManager", other: "KlinkerBlockManager"
) -> "KlinkerBlockManager":
    """Combine blocks.

    Args:
      this: one block manager to combine
      other: other block manager to combine

    Returns:
      Combined KlinkerBlockManager

    Examples:

        >>> from klinker import KlinkerBlockManager
        >>> kbm = KlinkerBlockManager.from_dict({"block1": [[1,3,4],[3,4,5]], "block2": [[3,4,5],[5,6]]}, dataset_names=("A","B"))
        >>> kbm2 = KlinkerBlockManager.from_dict({"block3": [[7,4],[12,8]]}, dataset_names=("A","B"))
        >>> kbm_merged = KlinkerBlockManager.combine(kbm, kbm2)
        >>> kbm_merged.blocks.compute()
                        A          B
        block1  [1, 3, 4]  [3, 4, 5]
        block2  [3, 4, 5]     [5, 6]
        block3     [7, 4]    [12, 8]

    """

    def _merge_blocks(
        row: pd.Series, output_names: Sequence[str], left_right_names: Sequence[str]
    ):
        nonnull = row[~row.isnull()]
        if len(nonnull) == 2:  # no block overlap
            nonnull.index = output_names
            return nonnull
        else:
            A_left = set(nonnull[left_right_names[0]])
            A_right = set(nonnull[left_right_names[2]])
            B_left = set(nonnull[left_right_names[1]])
            B_right = set(nonnull[left_right_names[3]])
            A = list(A_left.union(A_right))
            B = list(B_left.union(B_right))
            return pd.Series([A, B], index=output_names, name=nonnull.name)

    if list(this.blocks.columns) != list(other.blocks.columns):
        raise ValueError("Cannot combine blocks from different datasets!")

    output_names = this.blocks.columns
    left_suffix = "left"
    right_suffix = "right"
    left_right_names = [
        col + suffix
        for col_names, suffix in zip(
            [this.blocks.columns, other.blocks.columns], [left_suffix, right_suffix]
        )
        for col in col_names
    ]
    joined = this.blocks.join(
        other.blocks, how="outer", lsuffix="left", rsuffix="right"
    )

    meta = pd.DataFrame([], columns=output_names)
    return cls(
        joined.apply(
            _merge_blocks,
            output_names=output_names,
            left_right_names=left_right_names,
            axis=1,
            meta=meta,
        )
    )

entity_pairs(entity_id, column_id)

Get all pairs where this entity shows up.

Parameters:

Name Type Description Default
entity_id Union[str, int]

Union[str, int]: Entity id.

required
column_id int

int: Whether entity belongs to left (0) or right (1) dataset.

required

Returns:

Type Description
Generator[Tuple[Union[int, str], ...], None, None]

Generator for these pairs.

Source code in klinker/data/blocks.py
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
def entity_pairs(
    self, entity_id: Union[str, int], column_id: int
) -> Generator[Tuple[Union[int, str], ...], None, None]:
    """Get all pairs where this entity shows up.

    Args:
      entity_id: Union[str, int]: Entity id.
      column_id: int: Whether entity belongs to left (0) or right (1) dataset.

    Returns:
        Generator for these pairs.
    """
    cur_blocks = self.find_blocks(entity_id, column_id)
    other_column = 0 if column_id == 1 else 1
    other_column_name = self.blocks.columns[other_column]
    return (
        pair
        for blk_name in cur_blocks
        for _, blk in self.blocks.loc[blk_name][other_column_name].compute().items()
        for pair in itertools.product({entity_id}, blk)
    )

find_blocks(entity_id, column_id)

Find blocks where entity id belongs to.

Parameters:

Name Type Description Default
entity_id Union[str, int]

Union[str, int]: Entity id.

required
column_id int

int: Whether entity belongs to left (0) or right (1) dataset.

required

Returns:

Type Description
ndarray

Blocks where entity id belongs to.

Source code in klinker/data/blocks.py
388
389
390
391
392
393
394
395
396
397
398
def find_blocks(self, entity_id: Union[str, int], column_id: int) -> np.ndarray:
    """Find blocks where entity id belongs to.

    Args:
      entity_id: Union[str, int]: Entity id.
      column_id: int: Whether entity belongs to left (0) or right (1) dataset.

    Returns:
        Blocks where entity id belongs to.
    """
    return self._grouped[column_id].get_group(entity_id).index.values.compute()

from_dict(block_dict, dataset_names=('left', 'right'), npartitions=1, **kwargs) classmethod

Parameters:

Name Type Description Default
block_dict Dict[BlockIdTypeVar, Tuple[List[EntityIdTypeVar], List[EntityIdTypeVar]]]

Dictionary with block information.

required
dataset_names Tuple[str, str]

Tuple[str, str]: Tuple of dataset names.

('left', 'right')
npartitions int

int: Partitions used for dask.

1
**kwargs

Passed to dd.from_dict.

{}

Returns:

Type Description
KlinkerBlockManager

Blocks as KlinkerBlockManager

Examples:

>>> from klinker import KlinkerBlockManager
>>> kbm = KlinkerBlockManager.from_dict({"block1": [[1,3,4],[3,4,5]], "block2": [[3,4,5],[5,6]]}, dataset_names=("A","B"))
Source code in klinker/data/blocks.py
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
@classmethod
def from_dict(
    cls,
    block_dict: Dict[
        BlockIdTypeVar, Tuple[List[EntityIdTypeVar], List[EntityIdTypeVar]]
    ],
    dataset_names: Tuple[str, str] = ("left", "right"),
    npartitions: int = 1,
    **kwargs,
) -> "KlinkerBlockManager":
    """

    Args:
      block_dict: Dictionary with block information.
      dataset_names: Tuple[str, str]: Tuple of dataset names.
      npartitions: int: Partitions used for dask.
      **kwargs: Passed to `dd.from_dict`.

    Returns:
        Blocks as KlinkerBlockManager

    Examples:

        >>> from klinker import KlinkerBlockManager
        >>> kbm = KlinkerBlockManager.from_dict({"block1": [[1,3,4],[3,4,5]], "block2": [[3,4,5],[5,6]]}, dataset_names=("A","B"))

    """
    return cls(
        dd.from_dict(
            block_dict,
            orient="index",
            columns=dataset_names,
            npartitions=npartitions,
            **kwargs,
        )
    )

from_pandas(df, npartitions=1, **kwargs) classmethod

Create from pandas.

Parameters:

Name Type Description Default
df DataFrame

pd.DataFrame: DataFrame

required
npartitions int

int: Partitions for dask

1
**kwargs

Passed to dd.from_pandas

{}

Returns:

Type Description
KlinkerBlockManager

Blocks as KlinkerBlockManager

Examples:

>>> import pandas as pd
>>> from klinker import KlinkerBlockManager
>>> pd_blocks = pd.DataFrame({'A': {'block1': [1, 3, 4], 'block2': [3, 4, 5]}, 'B': {'block1': [3, 4, 5], 'block2': [5, 6]}})
>>> kbm = KlinkerBlockManager.from_pandas(pd_blocks)
Source code in klinker/data/blocks.py
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
@classmethod
def from_pandas(
    cls, df: pd.DataFrame, npartitions: int = 1, **kwargs
) -> "KlinkerBlockManager":
    """Create from pandas.

    Args:
      df: pd.DataFrame: DataFrame
      npartitions: int:  Partitions for dask
      **kwargs: Passed to `dd.from_pandas`

    Returns:
        Blocks as KlinkerBlockManager

    Examples:

        >>> import pandas as pd
        >>> from klinker import KlinkerBlockManager
        >>> pd_blocks = pd.DataFrame({'A': {'block1': [1, 3, 4], 'block2': [3, 4, 5]}, 'B': {'block1': [3, 4, 5], 'block2': [5, 6]}})
        >>> kbm = KlinkerBlockManager.from_pandas(pd_blocks)

    """
    return cls(dd.from_pandas(df, npartitions=npartitions, **kwargs))

read_parquet(path, calculate_divisions=True, **kwargs) classmethod

Read blocks from parquet.

Parameters:

Name Type Description Default
path Union[str, Path]

Union[str, pathlib.Path]: Path where blocks are stored.

required
calculate_divisions bool

bool: Calculate index divisions.

True
**kwargs

Passed to dd.read_parquet function.

{}

Returns:

Type Description
KlinkerBlockManager

Blocks as KlinkerBlockManager

Source code in klinker/data/blocks.py
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
@classmethod
def read_parquet(
    cls,
    path: Union[str, pathlib.Path],
    calculate_divisions: bool = True,
    **kwargs,
) -> "KlinkerBlockManager":
    """Read blocks from parquet.

    Args:
      path: Union[str, pathlib.Path]: Path where blocks are stored.
      calculate_divisions: bool: Calculate index divisions.
      **kwargs: Passed to `dd.read_parquet` function.

    Returns:
        Blocks as KlinkerBlockManager
    """
    return cls(
        dd.read_parquet(
            path=path,
            calculate_divisions=calculate_divisions,
            **kwargs,
        )
    )

to_dict()

Return blocks as dict.

Returns:

Type Description
Dict[Union[str, int], Tuple[Union[str, int], Union[str, int]]]

The dict has block names as keys and a tuple of sets of entity ids.

Source code in klinker/data/blocks.py
376
377
378
379
380
381
382
383
384
385
386
def to_dict(self) -> Dict[Union[str, int], Tuple[Union[str, int], Union[str, int]]]:
    """Return blocks as dict.

    Returns:
      The dict has block names as keys and a tuple of sets of entity ids.
    """
    return (
        self.blocks.apply(tuple, axis=1, meta=pd.Series([], dtype=object))
        .compute()
        .to_dict()
    )

to_parquet(path, **kwargs)

Write blocks as parquet file(s).

Parameters:

Name Type Description Default
path Union[str, Path]

Union[str, pathlib.Path]: Where to write.

required
**kwargs

passed to the parquet function

{}
Source code in klinker/data/blocks.py
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
def to_parquet(self, path: Union[str, pathlib.Path], **kwargs):
    """Write blocks as parquet file(s).

    Args:
      path: Union[str, pathlib.Path]: Where to write.
      **kwargs: passed to the parquet function
    """
    if "schema" not in kwargs:
        left, right = self.blocks.columns[:2]
        block_type = pa.list_(pa.string())
        schema = {
            left: block_type,
            right: block_type,
        }
    else:
        schema = kwargs.pop["schema"]  # type: ignore
    try:
        self.blocks.to_parquet(path, schema=schema, **kwargs)
    except ValueError:
        # If index is incorrectly assumed by dask to be string
        # and it turns out to be int64 an error would be thrown
        # This is kind of a dirty hack
        schema["__null_dask_index__"] = pa.int64()
        self.blocks.to_parquet(path, schema=schema, **kwargs)

KlinkerDaskFrame

Bases: DataFrame, AbstractKlinkerFrame

Parallel KlinkerFrame.

Please don't use the __init__ method but rather from_dask_dataframe for initialisation!

Parameters:

Name Type Description Default
dsk

The dask graph to compute this KlinkerFrame

required
name

The key prefix that specifies which keys in the dask comprise this particular KlinkerFrame

required
meta

An empty klinkerframe object with names, dtypes, and indices matching the expected output.

required
divisions

Values along which we partition our blocks on the index

required

Returns:

Type Description

KlinkerDaskFrame

Examples:

>>> import pandas as pd
>>> from klinker.data import KlinkerDaskFrame
>>> import dask.dataframe as dd
>>> df = dd.from_pandas(pd.DataFrame([("1","John", "Doe"),("2","Jane","Doe")],columns=["id","first name", "surname"]),npartitions=1)
>>> kdf = KlinkerDaskFrame.from_dask_dataframe(df, table_name="A", id_col="id")
>>> kdf
Dask KlinkerDaskFrame Structure:
                   id first name surname
npartitions=1
0              object     object  object
1                 ...        ...     ...
Dask Name: KlinkerPandasFrame, 2 graph layers
Table Name: A, id_col: id
Source code in klinker/data/enhanced_df.py
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
class KlinkerDaskFrame(dd.core.DataFrame, AbstractKlinkerFrame):
    """Parallel KlinkerFrame.

    Please don't use the `__init__` method but rather `from_dask_dataframe` for
    initialisation!

    Args:
      dsk: The dask graph to compute this KlinkerFrame
      name: The key prefix that specifies which keys in the dask comprise this particular KlinkerFrame
      meta: An empty klinkerframe object with names, dtypes, and indices matching the expected output.
      divisions: Values along which we partition our blocks on the index

    Returns:
        KlinkerDaskFrame

    Examples:

        >>> import pandas as pd
        >>> from klinker.data import KlinkerDaskFrame
        >>> import dask.dataframe as dd
        >>> df = dd.from_pandas(pd.DataFrame([("1","John", "Doe"),("2","Jane","Doe")],columns=["id","first name", "surname"]),npartitions=1)
        >>> kdf = KlinkerDaskFrame.from_dask_dataframe(df, table_name="A", id_col="id")
        >>> kdf
        Dask KlinkerDaskFrame Structure:
                           id first name surname
        npartitions=1
        0              object     object  object
        1                 ...        ...     ...
        Dask Name: KlinkerPandasFrame, 2 graph layers
        Table Name: A, id_col: id

    """

    _partition_type = KlinkerPandasFrame

    def __init__(
        self,
        dsk,
        name,
        meta,
        divisions,
        table_name: Optional[str] = None,
        id_col: str = "id",
    ):
        super().__init__(dsk, name, meta, divisions)
        if table_name is None:
            self._table_name = meta.table_name
            self._id_col = meta.id_col
        else:
            self._table_name = table_name
            self._id_col = id_col

    @staticmethod
    def _static_propagate_klinker_attributes(
        new_object: "KlinkerDaskFrame", table_name: str, id_col: str
    ) -> "KlinkerDaskFrame":
        new_object.table_name = table_name
        new_object.id_col = id_col
        return new_object

    @property
    def non_id_columns(self) -> List[str]:
        """All columns which are not `id_col`"""
        return self._meta.non_id_columns

    @classmethod
    def _upgrade_from_series(
        cls,
        series,
        columns: List[str],
        table_name: Optional[str],
        id_col: str,
        reset_index: bool = True,
        meta=no_default,
    ) -> "KlinkerFrame":
        assert table_name
        kf = series.map_partitions(
            KlinkerPandasFrame._upgrade_from_series,
            columns=columns,
            table_name=table_name,
            id_col=id_col,
            reset_index=reset_index,
            meta=meta,
        )
        return KlinkerDaskFrame._static_propagate_klinker_attributes(
            kf, table_name, id_col
        )

    def concat_values(
        self,
    ) -> dd.Series:
        """Concatenate attribute values.

        Returns:
            dd.Series with concatenated values and id_col as index.

        Examples:

            >>> import pandas as pd
            >>> from klinker.data import KlinkerDaskFrame
            >>> import dask.dataframe as dd
            >>> df = dd.from_pandas(pd.DataFrame([("1","John", "Doe"),("2","Jane","Doe")],columns=["id","first name", "surname"]),npartitions=1)
            >>> kdf = KlinkerDaskFrame.from_dask_dataframe(df, table_name="A", id_col="id")
            >>> kdf.concat_values().compute()
            id
            1    John Doe
            2    Jane Doe
            Name: A, dtype: object

        """
        self = self.fillna("")
        assert self.table_name
        meta = pd.Series([], name=self.table_name, dtype="str")
        meta.index.name = self.id_col
        return self.map_partitions(
            M.concat_values,
            meta=meta,
        )

    @classmethod
    def from_dask_dataframe(
        cls,
        df: dd.DataFrame,
        table_name: str,
        id_col: str,
        meta=no_default,
        construction_class: Type[KlinkerPandasFrame] = KlinkerPandasFrame,
    ) -> "KlinkerDaskFrame":
        """Create KlinkDaskFrame from dask dataframe.

        Args:
          df: dd.DataFrame: Dask dataframe.
          table_name: str: Name of dataset.
          id_col: str: Column where entity_ids are stored
          meta: meta for dask
          construction_class: Either :class:`KlinkerPandasFrame` or :class:`KlinkerTriplePandasFrame`

        Returns:
            KlinkerDaskFrame

        Examples:

            >>> import pandas as pd
            >>> from klinker.data import KlinkerDaskFrame
            >>> import dask.dataframe as dd
            >>> df = dd.from_pandas(pd.DataFrame([("1","John", "Doe"),("2","Jane","Doe")],columns=["id","first name", "surname"]),npartitions=1)
            >>> kdf = KlinkerDaskFrame.from_dask_dataframe(df, table_name="A", id_col="id")
            >>> kdf
            Dask KlinkerDaskFrame Structure:
                               id first name surname
            npartitions=1
            0              object     object  object
            1                 ...        ...     ...
            Dask Name: KlinkerPandasFrame, 2 graph layers
            Table Name: A, id_col: id

        """
        new_df = df.map_partitions(
            construction_class,
            table_name=table_name,
            id_col=id_col,
            meta=meta,
        )
        meta = new_df._meta if meta is no_default else meta
        return cls(
            dsk=new_df.dask,
            name=new_df._name,
            meta=meta,
            divisions=new_df.divisions,
            table_name=table_name,
            id_col=id_col,
        )

    def __repr__(self) -> str:
        return (
            super().__repr__()
            + f"\nTable Name: {self.table_name}, id_col: {self.id_col}"
        )

non_id_columns: List[str] property

All columns which are not id_col

concat_values()

Concatenate attribute values.

Returns:

Type Description
Series

dd.Series with concatenated values and id_col as index.

Examples:

>>> import pandas as pd
>>> from klinker.data import KlinkerDaskFrame
>>> import dask.dataframe as dd
>>> df = dd.from_pandas(pd.DataFrame([("1","John", "Doe"),("2","Jane","Doe")],columns=["id","first name", "surname"]),npartitions=1)
>>> kdf = KlinkerDaskFrame.from_dask_dataframe(df, table_name="A", id_col="id")
>>> kdf.concat_values().compute()
id
1    John Doe
2    Jane Doe
Name: A, dtype: object
Source code in klinker/data/enhanced_df.py
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
def concat_values(
    self,
) -> dd.Series:
    """Concatenate attribute values.

    Returns:
        dd.Series with concatenated values and id_col as index.

    Examples:

        >>> import pandas as pd
        >>> from klinker.data import KlinkerDaskFrame
        >>> import dask.dataframe as dd
        >>> df = dd.from_pandas(pd.DataFrame([("1","John", "Doe"),("2","Jane","Doe")],columns=["id","first name", "surname"]),npartitions=1)
        >>> kdf = KlinkerDaskFrame.from_dask_dataframe(df, table_name="A", id_col="id")
        >>> kdf.concat_values().compute()
        id
        1    John Doe
        2    Jane Doe
        Name: A, dtype: object

    """
    self = self.fillna("")
    assert self.table_name
    meta = pd.Series([], name=self.table_name, dtype="str")
    meta.index.name = self.id_col
    return self.map_partitions(
        M.concat_values,
        meta=meta,
    )

from_dask_dataframe(df, table_name, id_col, meta=no_default, construction_class=KlinkerPandasFrame) classmethod

Create KlinkDaskFrame from dask dataframe.

Parameters:

Name Type Description Default
df DataFrame

dd.DataFrame: Dask dataframe.

required
table_name str

str: Name of dataset.

required
id_col str

str: Column where entity_ids are stored

required
meta

meta for dask

no_default
construction_class Type[KlinkerPandasFrame]

Either :class:KlinkerPandasFrame or :class:KlinkerTriplePandasFrame

KlinkerPandasFrame

Returns:

Type Description
KlinkerDaskFrame

KlinkerDaskFrame

Examples:

>>> import pandas as pd
>>> from klinker.data import KlinkerDaskFrame
>>> import dask.dataframe as dd
>>> df = dd.from_pandas(pd.DataFrame([("1","John", "Doe"),("2","Jane","Doe")],columns=["id","first name", "surname"]),npartitions=1)
>>> kdf = KlinkerDaskFrame.from_dask_dataframe(df, table_name="A", id_col="id")
>>> kdf
Dask KlinkerDaskFrame Structure:
                   id first name surname
npartitions=1
0              object     object  object
1                 ...        ...     ...
Dask Name: KlinkerPandasFrame, 2 graph layers
Table Name: A, id_col: id
Source code in klinker/data/enhanced_df.py
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
@classmethod
def from_dask_dataframe(
    cls,
    df: dd.DataFrame,
    table_name: str,
    id_col: str,
    meta=no_default,
    construction_class: Type[KlinkerPandasFrame] = KlinkerPandasFrame,
) -> "KlinkerDaskFrame":
    """Create KlinkDaskFrame from dask dataframe.

    Args:
      df: dd.DataFrame: Dask dataframe.
      table_name: str: Name of dataset.
      id_col: str: Column where entity_ids are stored
      meta: meta for dask
      construction_class: Either :class:`KlinkerPandasFrame` or :class:`KlinkerTriplePandasFrame`

    Returns:
        KlinkerDaskFrame

    Examples:

        >>> import pandas as pd
        >>> from klinker.data import KlinkerDaskFrame
        >>> import dask.dataframe as dd
        >>> df = dd.from_pandas(pd.DataFrame([("1","John", "Doe"),("2","Jane","Doe")],columns=["id","first name", "surname"]),npartitions=1)
        >>> kdf = KlinkerDaskFrame.from_dask_dataframe(df, table_name="A", id_col="id")
        >>> kdf
        Dask KlinkerDaskFrame Structure:
                           id first name surname
        npartitions=1
        0              object     object  object
        1                 ...        ...     ...
        Dask Name: KlinkerPandasFrame, 2 graph layers
        Table Name: A, id_col: id

    """
    new_df = df.map_partitions(
        construction_class,
        table_name=table_name,
        id_col=id_col,
        meta=meta,
    )
    meta = new_df._meta if meta is no_default else meta
    return cls(
        dsk=new_df.dask,
        name=new_df._name,
        meta=meta,
        divisions=new_df.divisions,
        table_name=table_name,
        id_col=id_col,
    )

KlinkerDataset dataclass

Helper class to hold info of benchmark datasets.

Source code in klinker/data/ea_dataset.py
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@dataclass
class KlinkerDataset:
    """Helper class to hold info of benchmark datasets."""

    left: KlinkerFrame
    right: KlinkerFrame
    gold: pd.DataFrame
    left_rel: Optional[pd.DataFrame] = None
    right_rel: Optional[pd.DataFrame] = None

    @classmethod
    def from_sylloge(cls, dataset: EADataset, clean: bool = False) -> "KlinkerDataset":
        """Create a klinker dataset from sylloge dataset.

        Args:
          dataset: EADataset: Sylloge dataset.
          clean: bool: Clean attribute information.

        Returns:
            klinker dataset

        Examples:

            >>> # doctest: +SKIP
            >>> from sylloge import MovieGraphBenchmark
            >>> ds = KlinkerDataset.from_sylloge(MovieGraphBenchmark())

        """
        left: Union[KlinkerDaskFrame, KlinkerPandasFrame]
        right: Union[KlinkerDaskFrame, KlinkerPandasFrame]
        if dataset.backend == "pandas":
            left = KlinkerTriplePandasFrame.from_df(
                dataset.attr_triples_left, table_name="left", id_col="head"
            )
            right = KlinkerTriplePandasFrame.from_df(
                dataset.attr_triples_right, table_name="right", id_col="head"
            )
        elif dataset.backend == "dask":
            left = KlinkerTripleDaskFrame.from_dask_dataframe(
                dataset.attr_triples_left, table_name="left", id_col="head"
            )
            right = KlinkerTripleDaskFrame.from_dask_dataframe(
                dataset.attr_triples_right, table_name="right", id_col="head"
            )
        else:
            raise ValueError(f"Unknown dataset backend {dataset.backend}")

        if clean:
            # remove datatype
            left["tail"] = left["tail"].map(lambda x: str(x).split("^^")[0])
            right["tail"] = right["tail"].map(lambda x: str(x).split("^^")[0])

        return cls(
            left=left,
            right=right,
            left_rel=dataset.rel_triples_left,
            right_rel=dataset.rel_triples_right,
            gold=dataset.ent_links,
        )

    def _sample_side(
        self, sample: pd.DataFrame, side: Side
    ) -> Tuple[KlinkerFrame, Optional[pd.DataFrame]]:
        if side == "left":
            rel_df = self.left_rel
            attr_df = self.left
            sample_col = sample.columns[0]
        else:
            rel_df = self.right_rel
            attr_df = self.right
            sample_col = sample.columns[1]
        sampled_attr_df = attr_df[attr_df[attr_df.id_col].isin(sample[sample_col])]
        if rel_df is None:
            return sampled_attr_df, None
        return (
            sampled_attr_df,
            rel_df[
                rel_df["head"].isin(sample[sample_col])
                | rel_df["tail"].isin(sample[sample_col])
            ],
        )

    def sample(self, size: int) -> "KlinkerDataset":
        """Get a sample of the dataset.

        Note:
            Currently this only takes the first n entities of the gold standard.

        Args:
          size: int: size of the sample

        Returns:
            sampled klinker dataset

        Examples:

            >>> # doctest: +SKIP
            >>> from sylloge import MovieGraphBenchmark
            >>> ds = KlinkerDataset.from_sylloge(MovieGraphBenchmark())
            >>> sampled = ds.sample(10)

        """
        # TODO actually sample
        sample_ent_links = self.gold.iloc[:size]
        sample_left, sample_left_rel = self._sample_side(sample_ent_links, "left")
        sample_right, sample_right_rel = self._sample_side(sample_ent_links, "right")
        return KlinkerDataset(
            left=sample_left,
            right=sample_right,
            left_rel=sample_left_rel,
            right_rel=sample_right_rel,
            gold=sample_ent_links,
        )

from_sylloge(dataset, clean=False) classmethod

Create a klinker dataset from sylloge dataset.

Parameters:

Name Type Description Default
dataset EADataset

EADataset: Sylloge dataset.

required
clean bool

bool: Clean attribute information.

False

Returns:

Type Description
KlinkerDataset

klinker dataset

Examples:

>>> # doctest: +SKIP
>>> from sylloge import MovieGraphBenchmark
>>> ds = KlinkerDataset.from_sylloge(MovieGraphBenchmark())
Source code in klinker/data/ea_dataset.py
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@classmethod
def from_sylloge(cls, dataset: EADataset, clean: bool = False) -> "KlinkerDataset":
    """Create a klinker dataset from sylloge dataset.

    Args:
      dataset: EADataset: Sylloge dataset.
      clean: bool: Clean attribute information.

    Returns:
        klinker dataset

    Examples:

        >>> # doctest: +SKIP
        >>> from sylloge import MovieGraphBenchmark
        >>> ds = KlinkerDataset.from_sylloge(MovieGraphBenchmark())

    """
    left: Union[KlinkerDaskFrame, KlinkerPandasFrame]
    right: Union[KlinkerDaskFrame, KlinkerPandasFrame]
    if dataset.backend == "pandas":
        left = KlinkerTriplePandasFrame.from_df(
            dataset.attr_triples_left, table_name="left", id_col="head"
        )
        right = KlinkerTriplePandasFrame.from_df(
            dataset.attr_triples_right, table_name="right", id_col="head"
        )
    elif dataset.backend == "dask":
        left = KlinkerTripleDaskFrame.from_dask_dataframe(
            dataset.attr_triples_left, table_name="left", id_col="head"
        )
        right = KlinkerTripleDaskFrame.from_dask_dataframe(
            dataset.attr_triples_right, table_name="right", id_col="head"
        )
    else:
        raise ValueError(f"Unknown dataset backend {dataset.backend}")

    if clean:
        # remove datatype
        left["tail"] = left["tail"].map(lambda x: str(x).split("^^")[0])
        right["tail"] = right["tail"].map(lambda x: str(x).split("^^")[0])

    return cls(
        left=left,
        right=right,
        left_rel=dataset.rel_triples_left,
        right_rel=dataset.rel_triples_right,
        gold=dataset.ent_links,
    )

sample(size)

Get a sample of the dataset.

Note

Currently this only takes the first n entities of the gold standard.

Parameters:

Name Type Description Default
size int

int: size of the sample

required

Returns:

Type Description
KlinkerDataset

sampled klinker dataset

Examples:

>>> # doctest: +SKIP
>>> from sylloge import MovieGraphBenchmark
>>> ds = KlinkerDataset.from_sylloge(MovieGraphBenchmark())
>>> sampled = ds.sample(10)
Source code in klinker/data/ea_dataset.py
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def sample(self, size: int) -> "KlinkerDataset":
    """Get a sample of the dataset.

    Note:
        Currently this only takes the first n entities of the gold standard.

    Args:
      size: int: size of the sample

    Returns:
        sampled klinker dataset

    Examples:

        >>> # doctest: +SKIP
        >>> from sylloge import MovieGraphBenchmark
        >>> ds = KlinkerDataset.from_sylloge(MovieGraphBenchmark())
        >>> sampled = ds.sample(10)

    """
    # TODO actually sample
    sample_ent_links = self.gold.iloc[:size]
    sample_left, sample_left_rel = self._sample_side(sample_ent_links, "left")
    sample_right, sample_right_rel = self._sample_side(sample_ent_links, "right")
    return KlinkerDataset(
        left=sample_left,
        right=sample_right,
        left_rel=sample_left_rel,
        right_rel=sample_right_rel,
        gold=sample_ent_links,
    )

KlinkerPandasFrame

Bases: DataFrame, AbstractKlinkerFrame

Enhanced pandas Dataframe for klinker.

This keeps table_name and id_col as metadata throughout transformations as best as possible.

Furthermore specific methods for blocking are implemented.

Examples:

>>> import pandas as pd
>>> from klinker.data import KlinkerPandasFrame
>>> df = pd.DataFrame([("1","John", "Doe"),("2","Jane","Doe")],columns=["id","first name", "surname"])
>>> df
  id first name surname
0  1       John     Doe
1  2       Jane     Doe
>>> kdf = KlinkerPandasFrame.from_df(df, table_name="A", id_col="id")
>>> kdf
  id first name surname
0  1       John     Doe
1  2       Jane     Doe
Table Name: A, id_col: id
>>> kdf.non_id_columns
['first name', 'surname']
>>> kdf.concat_values()
id
1    John Doe
2    Jane Doe
Name: A, dtype: object
Source code in klinker/data/enhanced_df.py
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
class KlinkerPandasFrame(pd.DataFrame, AbstractKlinkerFrame):
    """Enhanced pandas Dataframe for klinker.

    This keeps `table_name` and `id_col` as metadata
    throughout transformations as best as possible.

    Furthermore specific methods for blocking are implemented.

    Examples:

        >>> import pandas as pd
        >>> from klinker.data import KlinkerPandasFrame
        >>> df = pd.DataFrame([("1","John", "Doe"),("2","Jane","Doe")],columns=["id","first name", "surname"])
        >>> df
          id first name surname
        0  1       John     Doe
        1  2       Jane     Doe
        >>> kdf = KlinkerPandasFrame.from_df(df, table_name="A", id_col="id")
        >>> kdf
          id first name surname
        0  1       John     Doe
        1  2       Jane     Doe
        Table Name: A, id_col: id
        >>> kdf.non_id_columns
        ['first name', 'surname']
        >>> kdf.concat_values()
        id
        1    John Doe
        2    Jane Doe
        Name: A, dtype: object

    """

    _metadata = ["_table_name", "_id_col"]

    def __init__(
        self,
        data=None,
        index: Optional[Axes] = None,
        columns: Optional[Axes] = None,
        dtype: Optional[Dtype] = None,
        copy: Optional[bool] = None,
        table_name: Optional[str] = None,
        id_col: Optional[str] = "id",
    ) -> None:
        super().__init__(
            data=data, index=index, columns=columns, dtype=dtype, copy=copy
        )
        assert id_col
        self._table_name = table_name
        self._id_col: str = id_col

    @property
    def _constructor(self):
        """ """
        return KlinkerPandasFrame

    @property
    def non_id_columns(self) -> List[str]:
        """ """
        return [c for c in self.columns if not c == self.id_col]

    @classmethod
    def from_df(
        cls, df: pd.DataFrame, table_name: str, id_col: Optional[str] = "id"
    ) -> "KlinkerPandasFrame":
        """Construct a KlinkerPandasFrame from a pd.DataFrame.

        Args:
          df: pd.DataFrame: The df holding the data
          table_name: str: Name of the dataset.
          id_col: Optional[str]:  Column with entity ids ("id" as default).

        Returns:
            KlinkerPandasFrame

        Examples:

            >>> import pandas as pd
            >>> from klinker.data import KlinkerPandasFrame
            >>> df = pd.DataFrame([("1","John", "Doe"),("2","Jane","Doe")],columns=["id","first name", "surname"])
            >>> kdf = KlinkerPandasFrame.from_df(df, table_name="A", id_col="id")
            >>> kdf
              id first name surname
            0  1       John     Doe
            1  2       Jane     Doe
            Table Name: A, id_col: id

        """
        return cls(data=df, table_name=table_name, id_col=id_col)

    def concat_values(
        self,
    ) -> pd.Series:
        """Concatenate all values, that are not in the id_col.

        Returns:
            Series with id_col as index and concatenated values.

        Examples:

            >>> import pandas as pd
            >>> from klinker.data import KlinkerPandasFrame
            >>> df = pd.DataFrame([("1","John", "Doe"),("2","Jane","Doe")],columns=["id","first name", "surname"])
            >>> kdf = KlinkerPandasFrame.from_df(df, table_name="A", id_col="id")
            >>> kdf.concat_values()
            id
            1    John Doe
            2    Jane Doe
            Name: A, dtype: object

        """
        self = self.fillna("")
        result = (
            self.copy()
            .set_index(self.id_col)[self.non_id_columns]
            .astype(str)
            .agg(" ".join, axis=1)
            .str.strip()
        )
        result.name = self.table_name
        return result

    @classmethod
    def _upgrade_from_series(
        cls,
        series,
        columns: List[str],
        table_name: Optional[str],
        id_col: str,
        reset_index: bool = True,
    ) -> "KlinkerFrame":
        kf = KlinkerPandasFrame(series.to_frame(), table_name=table_name, id_col=id_col)
        if reset_index:
            kf = kf.reset_index()
        kf.columns = columns
        return kf

    def __repr__(self) -> str:
        return (
            super().__repr__()
            + f"\nTable Name: {self.table_name}, id_col: {self.id_col}"
        )

non_id_columns: List[str] property

concat_values()

Concatenate all values, that are not in the id_col.

Returns:

Type Description
Series

Series with id_col as index and concatenated values.

Examples:

>>> import pandas as pd
>>> from klinker.data import KlinkerPandasFrame
>>> df = pd.DataFrame([("1","John", "Doe"),("2","Jane","Doe")],columns=["id","first name", "surname"])
>>> kdf = KlinkerPandasFrame.from_df(df, table_name="A", id_col="id")
>>> kdf.concat_values()
id
1    John Doe
2    Jane Doe
Name: A, dtype: object
Source code in klinker/data/enhanced_df.py
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def concat_values(
    self,
) -> pd.Series:
    """Concatenate all values, that are not in the id_col.

    Returns:
        Series with id_col as index and concatenated values.

    Examples:

        >>> import pandas as pd
        >>> from klinker.data import KlinkerPandasFrame
        >>> df = pd.DataFrame([("1","John", "Doe"),("2","Jane","Doe")],columns=["id","first name", "surname"])
        >>> kdf = KlinkerPandasFrame.from_df(df, table_name="A", id_col="id")
        >>> kdf.concat_values()
        id
        1    John Doe
        2    Jane Doe
        Name: A, dtype: object

    """
    self = self.fillna("")
    result = (
        self.copy()
        .set_index(self.id_col)[self.non_id_columns]
        .astype(str)
        .agg(" ".join, axis=1)
        .str.strip()
    )
    result.name = self.table_name
    return result

from_df(df, table_name, id_col='id') classmethod

Construct a KlinkerPandasFrame from a pd.DataFrame.

Parameters:

Name Type Description Default
df DataFrame

pd.DataFrame: The df holding the data

required
table_name str

str: Name of the dataset.

required
id_col Optional[str]

Optional[str]: Column with entity ids ("id" as default).

'id'

Returns:

Type Description
KlinkerPandasFrame

KlinkerPandasFrame

Examples:

>>> import pandas as pd
>>> from klinker.data import KlinkerPandasFrame
>>> df = pd.DataFrame([("1","John", "Doe"),("2","Jane","Doe")],columns=["id","first name", "surname"])
>>> kdf = KlinkerPandasFrame.from_df(df, table_name="A", id_col="id")
>>> kdf
  id first name surname
0  1       John     Doe
1  2       Jane     Doe
Table Name: A, id_col: id
Source code in klinker/data/enhanced_df.py
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
@classmethod
def from_df(
    cls, df: pd.DataFrame, table_name: str, id_col: Optional[str] = "id"
) -> "KlinkerPandasFrame":
    """Construct a KlinkerPandasFrame from a pd.DataFrame.

    Args:
      df: pd.DataFrame: The df holding the data
      table_name: str: Name of the dataset.
      id_col: Optional[str]:  Column with entity ids ("id" as default).

    Returns:
        KlinkerPandasFrame

    Examples:

        >>> import pandas as pd
        >>> from klinker.data import KlinkerPandasFrame
        >>> df = pd.DataFrame([("1","John", "Doe"),("2","Jane","Doe")],columns=["id","first name", "surname"])
        >>> kdf = KlinkerPandasFrame.from_df(df, table_name="A", id_col="id")
        >>> kdf
          id first name surname
        0  1       John     Doe
        1  2       Jane     Doe
        Table Name: A, id_col: id

    """
    return cls(data=df, table_name=table_name, id_col=id_col)

KlinkerTripleDaskFrame

Bases: KlinkerDaskFrame

Parallel KlinkerTriplePandasFrame

Parameters:

Name Type Description Default
dsk

The dask graph to compute this KlinkerFrame

required
name

The key prefix that specifies which keys in the dask comprise this particular KlinkerFrame

required
meta

An empty klinkerframe object with names, dtypes, and indices matching the expected output.

required
divisions

Values along which we partition our blocks on the index

required

Returns:

Source code in klinker/data/enhanced_df.py
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
class KlinkerTripleDaskFrame(KlinkerDaskFrame):
    """Parallel KlinkerTriplePandasFrame

    Args:
      dsk: The dask graph to compute this KlinkerFrame
      name: The key prefix that specifies which keys in the dask comprise this particular KlinkerFrame
      meta: An empty klinkerframe object with names, dtypes, and indices matching the expected output.
      divisions: Values along which we partition our blocks on the index

    Returns:

    """

    _partition_type = KlinkerTriplePandasFrame

    def concat_values(
        self,
    ) -> dd.Series:
        """


        Returns:

        """
        self = self.fillna("")
        assert self.table_name
        result = self.groupby(self.id_col)[self.columns[2]].apply(
            lambda grp: " ".join(grp.astype(str)).strip(),
            meta=pd.Series([], name=self.columns[2], dtype="str"),
        )
        result.name = self.table_name
        result._meta.index.name = self.id_col
        return result

concat_values()

Returns:

Source code in klinker/data/enhanced_df.py
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
def concat_values(
    self,
) -> dd.Series:
    """


    Returns:

    """
    self = self.fillna("")
    assert self.table_name
    result = self.groupby(self.id_col)[self.columns[2]].apply(
        lambda grp: " ".join(grp.astype(str)).strip(),
        meta=pd.Series([], name=self.columns[2], dtype="str"),
    )
    result.name = self.table_name
    result._meta.index.name = self.id_col
    return result

KlinkerTriplePandasFrame

Bases: KlinkerPandasFrame

Class for holding triple information.

Examples:

>>> import pandas as pd
>>> from klinker.data import KlinkerTriplePandasFrame
>>> df = pd.DataFrame([("e1","foaf:givenname","John"),("e1","foaf:family_name", "Doe"), ("e2","rdfs:label","Jane Doe")],columns=["head","rel","tail"])
>>> from klinker.data import KlinkerTriplePandasFrame
>>> kdf = KlinkerTriplePandasFrame.from_df(df, table_name="A",id_col="head")
>>> kdf
head               rel      tail
0   e1    foaf:givenname      John
1   e1  foaf:family_name       Doe
2   e2        rdfs:label  Jane Doe
Table Name: A, id_col: head
>>> kdf.concat_values()
head
e1    John Doe
e2    Jane Doe
Name: A, dtype: object
Source code in klinker/data/enhanced_df.py
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
class KlinkerTriplePandasFrame(KlinkerPandasFrame):
    """Class for holding triple information.

    Examples:

        >>> import pandas as pd
        >>> from klinker.data import KlinkerTriplePandasFrame
        >>> df = pd.DataFrame([("e1","foaf:givenname","John"),("e1","foaf:family_name", "Doe"), ("e2","rdfs:label","Jane Doe")],columns=["head","rel","tail"])
        >>> from klinker.data import KlinkerTriplePandasFrame
        >>> kdf = KlinkerTriplePandasFrame.from_df(df, table_name="A",id_col="head")
        >>> kdf
        head               rel      tail
        0   e1    foaf:givenname      John
        1   e1  foaf:family_name       Doe
        2   e2        rdfs:label  Jane Doe
        Table Name: A, id_col: head
        >>> kdf.concat_values()
        head
        e1    John Doe
        e2    Jane Doe
        Name: A, dtype: object

    """

    @property
    def _constructor(self):
        """ """
        return KlinkerTriplePandasFrame

    @property
    def non_id_columns(self) -> List[str]:
        """Last column."""
        return [self.columns[2]]

    def concat_values(
        self,
    ) -> pd.Series:
        """Concatenate all values of the tail column.

        Returns:
            Series with id_col as index and concatenated values.

        Examples:
            >>> import pandas as pd
            >>> from klinker.data import KlinkerTriplePandasFrame
            >>> df = pd.DataFrame([("e1","foaf:givenname","John"),("e1","foaf:family_name", "Doe"), ("e2","rdfs:label","Jane Doe")],columns=["head","rel","tail"])
            >>> from klinker.data import KlinkerTriplePandasFrame
            >>> kdf = KlinkerTriplePandasFrame.from_df(df, table_name="A",id_col="head")
            >>> kdf.concat_values()
            head
            e1    John Doe
            e2    Jane Doe
            Name: A, dtype: object
        """
        assert self.table_name
        self = self.fillna("")
        res = (
            self[[self.id_col, self.columns[2]]]
            .groupby(self.id_col)
            .agg(lambda row: " ".join(row.astype(str).values).strip())[self.columns[2]]
        )
        res.name = self.table_name
        return res

non_id_columns: List[str] property

Last column.

concat_values()

Concatenate all values of the tail column.

Returns:

Type Description
Series

Series with id_col as index and concatenated values.

Examples:

>>> import pandas as pd
>>> from klinker.data import KlinkerTriplePandasFrame
>>> df = pd.DataFrame([("e1","foaf:givenname","John"),("e1","foaf:family_name", "Doe"), ("e2","rdfs:label","Jane Doe")],columns=["head","rel","tail"])
>>> from klinker.data import KlinkerTriplePandasFrame
>>> kdf = KlinkerTriplePandasFrame.from_df(df, table_name="A",id_col="head")
>>> kdf.concat_values()
head
e1    John Doe
e2    Jane Doe
Name: A, dtype: object
Source code in klinker/data/enhanced_df.py
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
def concat_values(
    self,
) -> pd.Series:
    """Concatenate all values of the tail column.

    Returns:
        Series with id_col as index and concatenated values.

    Examples:
        >>> import pandas as pd
        >>> from klinker.data import KlinkerTriplePandasFrame
        >>> df = pd.DataFrame([("e1","foaf:givenname","John"),("e1","foaf:family_name", "Doe"), ("e2","rdfs:label","Jane Doe")],columns=["head","rel","tail"])
        >>> from klinker.data import KlinkerTriplePandasFrame
        >>> kdf = KlinkerTriplePandasFrame.from_df(df, table_name="A",id_col="head")
        >>> kdf.concat_values()
        head
        e1    John Doe
        e2    Jane Doe
        Name: A, dtype: object
    """
    assert self.table_name
    self = self.fillna("")
    res = (
        self[[self.id_col, self.columns[2]]]
        .groupby(self.id_col)
        .agg(lambda row: " ".join(row.astype(str).values).strip())[self.columns[2]]
    )
    res.name = self.table_name
    return res

NamedVector

Bases: Generic[T]

Class for holding named embeddings.

Parameters:

Name Type Description Default
names Union[List[str], Dict[str, int]]

Names of embedding rows.

required
vectors T

Embeddings.

required

Examples:

>>> import numpy as np
>>> from klinker.data import NamedVector
>>> emb = np.random.rand(5,2)
>>> names = [f"e_{i}" for i in range(len(emb))]
>>> nv = NamedVector(names, emb)
>>> nv # doctest: +SKIP
NamedVector(0|"e_0": [0.8307803 , 0.48443226],
        1|"e_1": [0.4887014 , 0.94833375],
        2|"e_2": [0.5665754 , 0.03836854],
        3|"e_3": [0.16981838, 0.22991864],
        4|"e_4": [0.23473072, 0.11225217],
        dtype=float64)
>>> nv[0] # doctest: +SKIP
array([0.8307803 , 0.48443226])
>>> nv["e_0"]  # doctest: +SKIP
array([0.8307803 , 0.48443226])
Source code in klinker/data/named_vector.py
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
class NamedVector(Generic[T]):
    """Class for holding named embeddings.

    Args:
        names: Names of embedding rows.
        vectors: Embeddings.

    Examples:

        >>> import numpy as np
        >>> from klinker.data import NamedVector
        >>> emb = np.random.rand(5,2)
        >>> names = [f"e_{i}" for i in range(len(emb))]
        >>> nv = NamedVector(names, emb)
        >>> nv # doctest: +SKIP
        NamedVector(0|"e_0": [0.8307803 , 0.48443226],
                1|"e_1": [0.4887014 , 0.94833375],
                2|"e_2": [0.5665754 , 0.03836854],
                3|"e_3": [0.16981838, 0.22991864],
                4|"e_4": [0.23473072, 0.11225217],
                dtype=float64)
        >>> nv[0] # doctest: +SKIP
        array([0.8307803 , 0.48443226])
        >>> nv["e_0"]  # doctest: +SKIP
        array([0.8307803 , 0.48443226])
    """

    _vectors: T
    _names: pd.Series

    def __init__(self, names: Union[List[str], Dict[str, int]], vectors: T):
        if isinstance(names, dict):
            self._names = pd.Series(names)
        elif isinstance(names, pd.Series):
            self._names = names
        else:
            self.names = names
        if not all(isinstance(x, str) for x in self._names.index):
            raise ValueError("The names index must be of type `str`")
        if not (self._names.values == np.arange(len(self._names))).all():
            raise ValueError("Indices must be contiguous!")
        self.vectors = vectors

    def _validate(self, names: List[str], vectors: T):
        if not len(set(names)) == len(names):
            raise ValueError("Names must be unique!")
        if len(names) != len(vectors):
            raise ValueError(
                f"Names and vectors must have same length but got len(names)={len(names)} and len(vectors)={len(vectors)}"
            )

    @property
    def _tensor_lib(self) -> ModuleType:
        return np if isinstance(self.vectors, np.ndarray) else torch

    @property
    def names(self) -> List[str]:
        """List of names"""
        return self._names.index.tolist()

    @names.setter
    def names(self, new_names: List[str]):
        if hasattr(self, "_vectors"):
            self._validate(new_names, self.vectors)
        self._names = pd.Series(list(range(len(new_names))), index=new_names)

    @property
    def vectors(self) -> T:
        """Embeddings."""
        return self._vectors

    @vectors.setter
    def vectors(self, new_vectors: T):
        if hasattr(self, "_names"):
            self._validate(self.names, new_vectors)
        self._vectors = new_vectors

    @property
    def entity_id_mapping(self) -> Dict[str, int]:
        """Mapping of entity names to ids"""
        return self._names.to_dict()

    @property
    def id_entity_mapping(self) -> Dict[int, str]:
        """Mapping of vectors indices to entity names"""
        return pd.Series(self._names.index.values, index=self._names).to_dict()

    def _key_handling(
        self, key: Union[str, int, List[str], List[int], slice]
    ) -> Union[int, pd.Series, List[int], slice]:
        if isinstance(key, int):
            return key
        elif isinstance(key, str):
            return self._names.loc[key]
        elif isinstance(key, Sequence):
            if len(key) == 0:
                return slice(0, 0, 1)
            elif isinstance(key[0], str):
                return self._names.loc[key]
            else:
                return key
        elif isinstance(key, slice):
            return key
        else:
            raise ValueError(f"Unsupported key type {type(key)} of key {key}")

    def __getitem__(self, key: Union[str, int, List[str], List[int], slice]) -> T:
        vector_key = self._key_handling(key)
        return self.vectors[vector_key]

    def __setitem__(self, key: Union[str, int, List[str], List[int], slice], value: T):
        if not isinstance(value, np.ndarray) and not isinstance(value, torch.Tensor):
            raise ValueError(
                f"Cannot assign value(s) with type {type(value)}\n To change the names use the `names` variable directly."
            )
        vector_key = self._key_handling(key)
        self.vectors[vector_key] = value

    def __repr__(self) -> str:
        if not hasattr(self, "_vectors"):
            return "NamedVector(names=None, vectors=None)"
        "torch.Tensor" if isinstance(self.vectors, torch.Tensor) else "np.ndarray"
        str_repr = ""
        spacing = "\t    "
        for idx, name_arr_line in enumerate(zip(self.names, self.vectors)):
            name, arr_line = name_arr_line
            line = f'{idx}|"{name}": {_shorten_tensor_repr(arr_line)},\n'
            if str_repr == "":
                str_repr = f"NamedVector({line}"
            else:
                str_repr += f"{spacing}{line}"

        str_repr += f"{spacing}dtype={self.vectors.dtype})"
        return str_repr

    def __len__(self) -> int:
        return len(self.names)

    def __eq__(self, other) -> bool:
        if not isinstance(other, NamedVector):
            return False
        if not self._tensor_lib == other._tensor_lib:
            return False
        if not all(self._names.index == other._names.index):
            return False
        if not all(self._names.values == other._names.values):
            return False
        if not (self._vectors == other._vectors).all():
            return False
        return True

    def concat(self, other: "NamedVector") -> "NamedVector":
        """Concatenate two NamedVector objects.

        Args:
          other: "NamedVector": Other instances to append.

        Returns:
            Concatenated named vector.

        Examples:

            >>> import numpy as np
            >>> from klinker.data import NamedVector
            >>> emb = np.random.rand(5,2)
            >>> names = [f"e_{i}" for i in range(len(emb))]
            >>> nv = NamedVector(names, emb)
            >>> emb2 = np.random.rand(2,2)
            >>> nv2 = NamedVector(["e_5","e_6"],emb2)
            >>> nv.concat(nv2) # doctest: +SKIP
            NamedVector(0|"e_0": [0.8307803 , 0.48443226],
                    1|"e_1": [0.4887014 , 0.94833375],
                    2|"e_2": [0.5665754 , 0.03836854],
                    3|"e_3": [0.16981838, 0.22991864],
                    4|"e_4": [0.23473072, 0.11225217],
                    5|"e_5": [0.83645295, 0.78669799],
                    6|"e_6": [0.77684827, 0.4350322 ],
                    dtype=float64)
        """
        new_vectors = self._tensor_lib.concatenate([self.vectors, other.vectors])
        new_names = self.names + other.names
        return NamedVector(names=new_names, vectors=new_vectors)

    def subset(self, key: Union[str, List[str]]) -> "NamedVector":
        """Return a subset as new object instance.

        Args:
          key: Union[str, List[str]]: key(s) of subset

        Returns:
            Specified subset of this instance.

        >>> import numpy as np
        >>> from klinker.data import NamedVector
        >>> emb = np.random.rand(5,2)
        >>> names = [f"e_{i}" for i in range(len(emb))]
        >>> nv = NamedVector(names, emb)
        >>> nv.subset(["e_1","e_3"]) # doctest: +SKIP
        NamedVector(0|"e_1": [0.4887014 , 0.94833375],
                1|"e_3": [0.16981838, 0.22991864],
                dtype=float64)
        """
        if isinstance(key, str):
            key = [key]
        sub_names = self._names.loc[key]
        sub_vectors = self._vectors[sub_names]
        # need to cast to list to ensure contiguous ids
        return NamedVector(names=sub_names.index.tolist(), vectors=sub_vectors)

    def to_pickle(self, path):
        """Save as pickle.

        See: `read_pickle`

        Args:
          path: Path where to save.
        """
        with open(path, "wb") as file_handle:
            pickle.dump((self.names, self.vectors), file_handle)

    @classmethod
    def from_pickle(cls, path) -> "NamedVector":
        """Read from pickle.

        See: `to_pickle`

        Args:
          path: Path from where to load.

        Returns:
            Loaded named vector
        """
        with open(path, "rb") as file_handle:
            names, vectors = pickle.load(file_handle)
        return cls(names, vectors)

entity_id_mapping: Dict[str, int] property

Mapping of entity names to ids

id_entity_mapping: Dict[int, str] property

Mapping of vectors indices to entity names

names: List[str] property writable

List of names

vectors: T property writable

Embeddings.

concat(other)

Concatenate two NamedVector objects.

Parameters:

Name Type Description Default
other NamedVector

"NamedVector": Other instances to append.

required

Returns:

Type Description
NamedVector

Concatenated named vector.

Examples:

>>> import numpy as np
>>> from klinker.data import NamedVector
>>> emb = np.random.rand(5,2)
>>> names = [f"e_{i}" for i in range(len(emb))]
>>> nv = NamedVector(names, emb)
>>> emb2 = np.random.rand(2,2)
>>> nv2 = NamedVector(["e_5","e_6"],emb2)
>>> nv.concat(nv2) # doctest: +SKIP
NamedVector(0|"e_0": [0.8307803 , 0.48443226],
        1|"e_1": [0.4887014 , 0.94833375],
        2|"e_2": [0.5665754 , 0.03836854],
        3|"e_3": [0.16981838, 0.22991864],
        4|"e_4": [0.23473072, 0.11225217],
        5|"e_5": [0.83645295, 0.78669799],
        6|"e_6": [0.77684827, 0.4350322 ],
        dtype=float64)
Source code in klinker/data/named_vector.py
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def concat(self, other: "NamedVector") -> "NamedVector":
    """Concatenate two NamedVector objects.

    Args:
      other: "NamedVector": Other instances to append.

    Returns:
        Concatenated named vector.

    Examples:

        >>> import numpy as np
        >>> from klinker.data import NamedVector
        >>> emb = np.random.rand(5,2)
        >>> names = [f"e_{i}" for i in range(len(emb))]
        >>> nv = NamedVector(names, emb)
        >>> emb2 = np.random.rand(2,2)
        >>> nv2 = NamedVector(["e_5","e_6"],emb2)
        >>> nv.concat(nv2) # doctest: +SKIP
        NamedVector(0|"e_0": [0.8307803 , 0.48443226],
                1|"e_1": [0.4887014 , 0.94833375],
                2|"e_2": [0.5665754 , 0.03836854],
                3|"e_3": [0.16981838, 0.22991864],
                4|"e_4": [0.23473072, 0.11225217],
                5|"e_5": [0.83645295, 0.78669799],
                6|"e_6": [0.77684827, 0.4350322 ],
                dtype=float64)
    """
    new_vectors = self._tensor_lib.concatenate([self.vectors, other.vectors])
    new_names = self.names + other.names
    return NamedVector(names=new_names, vectors=new_vectors)

from_pickle(path) classmethod

Read from pickle.

See: to_pickle

Parameters:

Name Type Description Default
path

Path from where to load.

required

Returns:

Type Description
NamedVector

Loaded named vector

Source code in klinker/data/named_vector.py
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
@classmethod
def from_pickle(cls, path) -> "NamedVector":
    """Read from pickle.

    See: `to_pickle`

    Args:
      path: Path from where to load.

    Returns:
        Loaded named vector
    """
    with open(path, "rb") as file_handle:
        names, vectors = pickle.load(file_handle)
    return cls(names, vectors)

subset(key)

Return a subset as new object instance.

Parameters:

Name Type Description Default
key Union[str, List[str]]

Union[str, List[str]]: key(s) of subset

required

Returns:

Type Description
NamedVector

Specified subset of this instance.

import numpy as np from klinker.data import NamedVector emb = np.random.rand(5,2) names = [f"e_{i}" for i in range(len(emb))] nv = NamedVector(names, emb) nv.subset(["e_1","e_3"]) # doctest: +SKIP NamedVector(0|"e_1": [0.4887014 , 0.94833375], 1|"e_3": [0.16981838, 0.22991864], dtype=float64)

Source code in klinker/data/named_vector.py
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def subset(self, key: Union[str, List[str]]) -> "NamedVector":
    """Return a subset as new object instance.

    Args:
      key: Union[str, List[str]]: key(s) of subset

    Returns:
        Specified subset of this instance.

    >>> import numpy as np
    >>> from klinker.data import NamedVector
    >>> emb = np.random.rand(5,2)
    >>> names = [f"e_{i}" for i in range(len(emb))]
    >>> nv = NamedVector(names, emb)
    >>> nv.subset(["e_1","e_3"]) # doctest: +SKIP
    NamedVector(0|"e_1": [0.4887014 , 0.94833375],
            1|"e_3": [0.16981838, 0.22991864],
            dtype=float64)
    """
    if isinstance(key, str):
        key = [key]
    sub_names = self._names.loc[key]
    sub_vectors = self._vectors[sub_names]
    # need to cast to list to ensure contiguous ids
    return NamedVector(names=sub_names.index.tolist(), vectors=sub_vectors)

to_pickle(path)

Save as pickle.

See: read_pickle

Parameters:

Name Type Description Default
path

Path where to save.

required
Source code in klinker/data/named_vector.py
238
239
240
241
242
243
244
245
246
247
def to_pickle(self, path):
    """Save as pickle.

    See: `read_pickle`

    Args:
      path: Path where to save.
    """
    with open(path, "wb") as file_handle:
        pickle.dump((self.names, self.vectors), file_handle)

from_klinker_frame(kf, npartitions)

Create KlinkerDaskFrame from KlinkerPandasFrame.

Parameters:

Name Type Description Default
kf KlinkerPandasFrame

KlinkerPandasFrame: Input dataframe

required
npartitions int

int: Number of partitions for dask.

required

Returns:

Type Description
KlinkerDaskFrame

KlinkerDaskFrame

Source code in klinker/data/enhanced_df.py
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
def from_klinker_frame(kf: KlinkerPandasFrame, npartitions: int) -> "KlinkerDaskFrame":
    """Create KlinkerDaskFrame from KlinkerPandasFrame.

    Args:
      kf: KlinkerPandasFrame: Input dataframe
      npartitions: int: Number of partitions for dask.

    Returns:
        KlinkerDaskFrame
    """
    if not kf.table_name:
        raise ValueError("KlinkerFrame needs to have a table_name set!")
    cls = (
        KlinkerTripleDaskFrame
        if isinstance(kf, KlinkerTriplePandasFrame)
        else KlinkerDaskFrame
    )
    return cls.from_dask_dataframe(
        dd.from_pandas(kf, npartitions=npartitions),
        table_name=kf.table_name,
        id_col=kf.id_col,
        meta=kf.head(0),
        construction_class=kf.__class__,
    )

generic_upgrade_from_series(conc, reset_index=False)

Upgrade a series to KlinkerFrame.

This automatically determines the correct KlinkerFrame class based on the given series class.

Note

This will use the series name as the resulting dataset name. The series index is assumed to be the entity ids.

Parameters:

Name Type Description Default
conc SeriesType

SeriesType: Series to upgrade.

required
reset_index bool

bool: If True resets index.

False

Returns:

Type Description
KlinkerFrame

KlinkerFrame

Examples:

>>> import pandas as pd
>>> from klinker.data import generic_upgrade_from_series
>>> ser = pd.Series(["John Doe","Jane Doe"],name="A",index=["e1","e2"])
>>> ser
e1    John Doe
e2    Jane Doe
Name: A, dtype: object
>>> generic_upgrade_from_series(ser, reset_index=True)
   id    values
0  e1  John Doe
1  e2  Jane Doe
Table Name: A, id_col: id
Source code in klinker/data/enhanced_df.py
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
def generic_upgrade_from_series(
    conc: SeriesType, reset_index: bool = False
) -> KlinkerFrame:
    """Upgrade a series to KlinkerFrame.

    This automatically determines the correct KlinkerFrame class
    based on the given series class.

    Note:
        This will use the series name as the resulting dataset name.
        The series index is assumed to be the entity ids.

    Args:
      conc: SeriesType: Series to upgrade.
      reset_index: bool: If True resets index.

    Returns:
        KlinkerFrame

    Examples:

        >>> import pandas as pd
        >>> from klinker.data import generic_upgrade_from_series
        >>> ser = pd.Series(["John Doe","Jane Doe"],name="A",index=["e1","e2"])
        >>> ser
        e1    John Doe
        e2    Jane Doe
        Name: A, dtype: object
        >>> generic_upgrade_from_series(ser, reset_index=True)
           id    values
        0  e1  John Doe
        1  e2  Jane Doe
        Table Name: A, id_col: id

    """
    frame_class: Type[KlinkerFrame]
    id_col = "id"
    if isinstance(conc, pd.Series):
        frame_class = KlinkerPandasFrame
        if conc.index.name is None:
            conc.index.name = "id"
        else:
            id_col = conc.index.name
    else:
        frame_class = KlinkerDaskFrame
        if conc.index.name is None:
            conc._meta.index.name = "id"
        else:
            id_col = conc.index.name
    columns = ["values"] if not reset_index else [id_col, "values"]
    return frame_class._upgrade_from_series(
        conc,
        columns=columns,
        table_name=conc.name,
        id_col=id_col,
        reset_index=reset_index,
    )