Skip to content

deCODE manifest

gentropy.datasource.decode.manifest.deCODEManifest dataclass

Bases: Dataset

Catalogue of deCODE summary-statistics files derived from an S3 bucket listing.

Each row corresponds to one SomaScan assay and records the S3 path of its gzipped TSV summary-statistics file together with provenance metadata such as file size and accession timestamp. The studyId follows the convention {projectId}_{Proteomics_*} as embedded in the file path.

The manifest is produced by from_s3 and later consumed by deCODEStudyIndex and deCODESummaryStatistics.

Source code in src/gentropy/datasource/decode/manifest.py 
 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
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
class deCODEManifest(Dataset):
    """Catalogue of deCODE summary-statistics files derived from an S3 bucket listing.

    Each row corresponds to one SomaScan assay and records the S3 path of its
    gzipped TSV summary-statistics file together with provenance metadata such as
    file size and accession timestamp.  The `studyId` follows the convention
    ``{projectId}_{Proteomics_*}`` as embedded in the file path.

    The manifest is produced by `from_s3` and later consumed by
    `deCODEStudyIndex` and `deCODESummaryStatistics`.
    """

    @classmethod
    def get_schema(cls) -> t.StructType:
        """Return the enforced Spark schema for `deCODEManifest`.

        Returns:
            t.StructType: Schema with fields ``projectId``, ``studyId``,
                ``hasSumstats``, ``summarystatsLocation``, ``size``, and
                ``accessionTimestamp``.
        """
        return t.StructType(
            [
                t.StructField("projectId", t.StringType()),
                t.StructField("studyId", t.StringType()),
                t.StructField("hasSumstats", t.BooleanType()),
                t.StructField("summarystatsLocation", t.StringType()),
                t.StructField("size", t.StringType()),
                t.StructField("accessionTimestamp", t.TimestampType()),
            ]
        )

    def get_summary_statistics_paths(self) -> list[str]:
        """Get summary statistics paths from manifest.

        Returns:
            list[str]: List of summary statistics paths.

        Examples:
            >>> data = [("path1",), ("path2",)]
            >>> schema = "summarystatsLocation STRING"
            >>> df = spark.createDataFrame(data, schema)
            >>> manifest = deCODEManifest(df)
            >>> manifest.get_summary_statistics_paths()
            ['path1', 'path2']
        """
        return [
            row.summarystatsLocation
            for row in self.df.select("summarystatsLocation").collect()
        ]

    @classmethod
    def from_s3(
        cls,
        session: Session,
        bucket_name: str,
        prefix: str = "",
    ) -> deCODEManifest:
        """Create a `deCODEManifest` by listing the S3 bucket directly via the Hadoop FileSystem API.

        Credentials (``bucket_name``, ``access_key_id``, ``secret_access_key``,
        ``s3_host_url``, ``s3_host_port``) can be requested at
        https://www.decode.com/summarydata/ and must be supplied to the session via
        ``add_s3_connector=True`` and ``s3_configuration`` (see :class:`~gentropy.external.s3.S3Config`).
        Once the session is configured, the S3A endpoint and credentials are resolved
        automatically from Hadoop's configuration.

        Args:
            session (Session): Active Gentropy Spark session with S3 connector configured.
            bucket_name (str): S3 bucket name without any URI prefix.
            prefix (str): Optional key prefix to restrict the listing (e.g. ``"Proteomics/"``).

        Returns:
            deCODEManifest: Populated manifest dataset.

        Examples:
            >>> manifest = deCODEManifest.from_s3(  # doctest: +SKIP
            ...     session=session,
            ...     bucket_name="largescaleplasma-2023",
            ... )
        """
        jvm = session.spark.sparkContext._jvm
        hadoop_conf = session.spark.sparkContext._jsc.hadoopConfiguration()

        base_uri = f"s3a://{bucket_name}/{prefix}"
        fs = jvm.org.apache.hadoop.fs.FileSystem.get(
            jvm.java.net.URI.create(base_uri),
            hadoop_conf,
        )

        iterator = fs.listFiles(jvm.org.apache.hadoop.fs.Path(base_uri), True)
        rows: list[tuple[str, str, int]] = []
        while iterator.hasNext():
            status = iterator.next()
            file_path: str = status.getPath().toString()
            if "Proteomics" in file_path and file_path.endswith(".txt.gz"):
                rows.append(
                    (
                        file_path,
                        cls._format_size_bytes(int(status.getLen())),
                        int(status.getModificationTime()),
                    )
                )

        return cls._rows_to_manifest(session, rows)

    @classmethod
    def _rows_to_manifest(
        cls,
        session: Session,
        rows: list[tuple[str, str, int]],
    ) -> deCODEManifest:
        """Build a `deCODEManifest` from pre-collected file-listing rows.

        Args:
            session (Session): Active Gentropy Spark session.
            rows (list[tuple[str, str, int]]): Each tuple is
                ``(summarystatsLocation, size, modTimeMs)`` where ``size``
                is already formatted (e.g. ``"926.0 MiB"``) and
                ``modTimeMs`` is the epoch-millisecond modification time.

        Returns:
            deCODEManifest: Populated manifest dataset.

        Examples:
            >>> rows = [
            ...     (
            ...         "s3a://b/Proteomics_SMP_PC0_1_G_P_0001.txt.gz",
            ...         "1.0 GiB",
            ...         1653820048000,
            ...     )
            ... ]
            >>> manifest = deCODEManifest._rows_to_manifest(spark, rows)  # doctest: +SKIP
        """
        raw_schema = t.StructType(
            [
                t.StructField("summarystatsLocation", t.StringType()),
                t.StructField("size", t.StringType()),
                t.StructField("_modTimeMs", t.LongType()),
            ]
        )

        project_id = f.when(
            f.col("summarystatsLocation").contains("Proteomics_SMP_"),
            f.lit(deCODEDataSource.DECODE_PROTEOMICS_SMP.value),
        ).otherwise(deCODEDataSource.DECODE_PROTEOMICS_RAW.value)

        manifest_df = (
            session.spark.createDataFrame(rows, raw_schema)
            .withColumn("projectId", project_id)
            .withColumn(
                "studyId",
                f.concat_ws(
                    "_",
                    f.col("projectId"),
                    f.regexp_extract(
                        "summarystatsLocation",
                        r"^.*/(Proteomics_.*)\.txt\.gz$",
                        1,
                    ),
                ),
            )
            .withColumn("hasSumstats", f.lit(True))
            .withColumn(
                "accessionTimestamp",
                f.to_timestamp(f.col("_modTimeMs") / 1000),
            )
            .select(
                "projectId",
                "studyId",
                "hasSumstats",
                "summarystatsLocation",
                "size",
                "accessionTimestamp",
            )
        )
        return cls(_df=manifest_df)

    @staticmethod
    def _format_size_bytes(n: int) -> str:
        """Format a byte count as a human-readable string matching ``aws s3 ls --human-readable``.

        Args:
            n (int): File size in bytes.

        Returns:
            str: Human-readable size string, e.g. ``"926.0 MiB"``.

        Examples:
            >>> deCODEManifest._format_size_bytes(970_981_376)
            '926.0 MiB'
            >>> deCODEManifest._format_size_bytes(1_073_741_824)
            '1.0 GiB'
            >>> deCODEManifest._format_size_bytes(512)
            '512.0 B'
        """
        value = float(n)
        for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
            if value < 1024:
                return f"{value:.1f} {unit}"
            value /= 1024
        return f"{value:.1f} PiB"

from_s3(session: Session, bucket_name: str, prefix: str = '') -> deCODEManifest classmethod

Create a deCODEManifest by listing the S3 bucket directly via the Hadoop FileSystem API.

Credentials (bucket_name, access_key_id, secret_access_key, s3_host_url, s3_host_port) can be requested at https://www.decode.com/summarydata/ and must be supplied to the session via add_s3_connector=True and s3_configuration (see :class:~gentropy.external.s3.S3Config). Once the session is configured, the S3A endpoint and credentials are resolved automatically from Hadoop's configuration.

Parameters:

Name Type Description Default
session Session

Active Gentropy Spark session with S3 connector configured.

 required
bucket_name str

S3 bucket name without any URI prefix.

 required
prefix str

Optional key prefix to restrict the listing (e.g. "Proteomics/").

 ''

Returns:

Name Type Description
deCODEManifest deCODEManifest

Populated manifest dataset.

Examples:

>>> manifest = deCODEManifest.from_s3(
...     session=session,
...     bucket_name="largescaleplasma-2023",
... )
Source code in src/gentropy/datasource/decode/manifest.py 
 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
@classmethod
def from_s3(
    cls,
    session: Session,
    bucket_name: str,
    prefix: str = "",
) -> deCODEManifest:
    """Create a `deCODEManifest` by listing the S3 bucket directly via the Hadoop FileSystem API.

    Credentials (``bucket_name``, ``access_key_id``, ``secret_access_key``,
    ``s3_host_url``, ``s3_host_port``) can be requested at
    https://www.decode.com/summarydata/ and must be supplied to the session via
    ``add_s3_connector=True`` and ``s3_configuration`` (see :class:`~gentropy.external.s3.S3Config`).
    Once the session is configured, the S3A endpoint and credentials are resolved
    automatically from Hadoop's configuration.

    Args:
        session (Session): Active Gentropy Spark session with S3 connector configured.
        bucket_name (str): S3 bucket name without any URI prefix.
        prefix (str): Optional key prefix to restrict the listing (e.g. ``"Proteomics/"``).

    Returns:
        deCODEManifest: Populated manifest dataset.

    Examples:
        >>> manifest = deCODEManifest.from_s3(  # doctest: +SKIP
        ...     session=session,
        ...     bucket_name="largescaleplasma-2023",
        ... )
    """
    jvm = session.spark.sparkContext._jvm
    hadoop_conf = session.spark.sparkContext._jsc.hadoopConfiguration()

    base_uri = f"s3a://{bucket_name}/{prefix}"
    fs = jvm.org.apache.hadoop.fs.FileSystem.get(
        jvm.java.net.URI.create(base_uri),
        hadoop_conf,
    )

    iterator = fs.listFiles(jvm.org.apache.hadoop.fs.Path(base_uri), True)
    rows: list[tuple[str, str, int]] = []
    while iterator.hasNext():
        status = iterator.next()
        file_path: str = status.getPath().toString()
        if "Proteomics" in file_path and file_path.endswith(".txt.gz"):
            rows.append(
                (
                    file_path,
                    cls._format_size_bytes(int(status.getLen())),
                    int(status.getModificationTime()),
                )
            )

    return cls._rows_to_manifest(session, rows)

get_schema() -> t.StructType classmethod

Return the enforced Spark schema for deCODEManifest.

Returns:

Type Description
StructType

t.StructType: Schema with fields projectId, studyId, hasSumstats, summarystatsLocation, size, and accessionTimestamp.
Source code in src/gentropy/datasource/decode/manifest.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
@classmethod
def get_schema(cls) -> t.StructType:
    """Return the enforced Spark schema for `deCODEManifest`.

    Returns:
        t.StructType: Schema with fields ``projectId``, ``studyId``,
            ``hasSumstats``, ``summarystatsLocation``, ``size``, and
            ``accessionTimestamp``.
    """
    return t.StructType(
        [
            t.StructField("projectId", t.StringType()),
            t.StructField("studyId", t.StringType()),
            t.StructField("hasSumstats", t.BooleanType()),
            t.StructField("summarystatsLocation", t.StringType()),
            t.StructField("size", t.StringType()),
            t.StructField("accessionTimestamp", t.TimestampType()),
        ]
    )

get_summary_statistics_paths() -> list[str]

Get summary statistics paths from manifest.

Returns:

Type Description
list[str]

list[str]: List of summary statistics paths.

Examples:

>>> data = [("path1",), ("path2",)]
>>> schema = "summarystatsLocation STRING"
>>> df = spark.createDataFrame(data, schema)
>>> manifest = deCODEManifest(df)
>>> manifest.get_summary_statistics_paths()
['path1', 'path2']
Source code in src/gentropy/datasource/decode/manifest.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def get_summary_statistics_paths(self) -> list[str]:
    """Get summary statistics paths from manifest.

    Returns:
        list[str]: List of summary statistics paths.

    Examples:
        >>> data = [("path1",), ("path2",)]
        >>> schema = "summarystatsLocation STRING"
        >>> df = spark.createDataFrame(data, schema)
        >>> manifest = deCODEManifest(df)
        >>> manifest.get_summary_statistics_paths()
        ['path1', 'path2']
    """
    return [
        row.summarystatsLocation
        for row in self.df.select("summarystatsLocation").collect()
    ]
  • 2026-02-10
  • 2026-02-10
  • Contributors
    • project-defiant