import{s as Ul,n as hl,o as ol}from"../chunks/scheduler.d6170356.js";import{S as jl,i as ml,g as p,s as t,r as c,A as wl,h as M,f as a,c as e,j as rl,u as i,x as y,k as zs,y as ul,a as n,v as J,d as r,t as U,w as h}from"../chunks/index.fcd4cc08.js";import{C as o}from"../chunks/CodeBlock.7b16bdef.js";import{H as as,E as Tl}from"../chunks/EditOnGithub.da2b595c.js";function dl(_s){let j,ns,ss,ts,u,es,T,qs="Spark enables real-time, large-scale data processing in a distributed environment.",ps,d,Hs="In particular you can use <code>huggingface_hub</code> to access Hugging Face datasets repositories in PySpark",Ms,b,ys,I,xs="To be able to read and write to Hugging Face URLs (e.g. <code>hf://datasets/username/dataset/data.parquet</code>), you need to install the <code>huggingface_hub</code> library:",cs,C,is,B,Ls="You also need to install <code>pyarrow</code> to read/write Parquet / JSON / CSV / etc. files using the filesystem API provided by <code>huggingFace_hub</code>:",Js,G,rs,f,Us,g,$s="You need to authenticate to Hugging Face to read private/gated dataset repositories or to write to your dataset repositories.",hs,Z,Ds="You can use the CLI for example:",os,k,js,S,Ks="It’s also possible to provide your Hugging Face token with the <code>HF_TOKEN</code> environment variable or passing the <code>storage_options</code> parameter to helper functions below:",ms,R,ws,W,Ps='For more details about authentication, check out <a href="https://huggingface.co/docs/huggingface_hub/quick-start#authentication" rel="nofollow">this guide</a>.',us,X,Ts,A,Os="PySpark doesn’t have an official support for Hugging Face paths, so we provide a helper function to read datasets in a distributed manner.",ds,Q,sl="For example you can read Parquet files from Hugging Face in an optimized way using PyArrow by defining this <code>read_parquet</code> helper function:",bs,Y,Is,N,ll=`Here is how we can use this on the <a href="https://huggingface.co/datasets/BAAI/Infinity-Instruct" rel="nofollow">BAAI/Infinity-Instruct</a> dataset.
It is a gated repository, users have to accept the terms of use before accessing it.`,Cs,m,al='<img class="block dark:hidden" src="https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/hub/datasets-spark-infinity-instruct-7M-min.png"/> <img class="hidden dark:block" src="https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/hub/datasets-spark-infinity-instruct-7M-dark-min.png"/>',Bs,F,nl="We use the <code>read_parquet</code> function to read data from the dataset, compute the number of dialogue per language and filter the dataset.",Gs,V,tl="After logging-in to access the gated repository, we can run:",fs,E,gs,v,el=`To compute the number of dialogues per language we run this code.
The <code>columns</code> argument is useful to only load the data we need, since PySpark doesn’t enable predicate push-down in this case.
There is also a <code>filters</code> argument to only load data with values within a certain range.`,Zs,z,ks,_,pl="To filter the dataset and only keep dialogues in Chinese:",Ss,q,Rs,H,Ws,x,Ml="We also provide a helper function to write datasets in a distributed manner to a Hugging Face repository.",Xs,L,yl=`You can write a PySpark Dataframe to Hugging Face using this <code>write_parquet</code> helper function based on the <code>huggingface_hub</code> API.
In particular it uses the <code>preupload_lfs_files</code> utility to upload Parquet files in parallel in a distributed manner, and only commits the files once they’re all uploaded:`,As,$,Qs,D,cl='Here is how we can use this function to write the filtered version of the <a href="https://huggingface.co/datasets/BAAI/Infinity-Instruct" rel="nofollow">BAAI/Infinity-Instruct</a> dataset back to Hugging Face.',Ys,K,il=`First you need to <a href="https://huggingface.co/new-dataset" rel="nofollow">create a dataset repository</a>, e.g. <code>username/Infinity-Instruct-Chinese-Only</code> (you can set it to private if you want).
Then, make sure you are authenticated and you can run:`,Ns,P,Fs,w,Jl='<img class="block dark:hidden" src="https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/hub/datasets-spark-infinity-instruct-chinese-only-min.png"/> <img class="hidden dark:block" src="https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/hub/datasets-spark-infinity-instruct-chinese-only-dark-min.png"/>',Vs,O,Es,ls,vs;return u=new as({props:{title:"Spark",local:"spark",headingTag:"h1"}}),b=new as({props:{title:"Installation",local:"installation",headingTag:"h2"}}),C=new o({props:{code:"cGlwJTIwaW5zdGFsbCUyMGh1Z2dpbmdmYWNlX2h1Yg==",highlighted:'pip <span class="hljs-keyword">install</span> huggingface_hub',wrap:!1}}),G=new o({props:{code:"cGlwJTIwaW5zdGFsbCUyMHB5YXJyb3c=",highlighted:'pip <span class="hljs-keyword">install</span> pyarrow',wrap:!1}}),f=new as({props:{title:"Authentication",local:"authentication",headingTag:"h2"}}),k=new o({props:{code:"aHVnZ2luZ2ZhY2UtY2xpJTIwbG9naW4=",highlighted:'huggingface-<span class="hljs-keyword">cli</span> login',wrap:!1}}),R=new o({props:{code:"c3RvcmFnZV9vcHRpb25zJTIwJTNEJTIwJTdCJTIydG9rZW4lMjIlM0ElMjAlMjJoZl94eHglMjIlN0Q=",highlighted:'storage_options = {<span class="hljs-string">&quot;token&quot;</span>: <span class="hljs-string">&quot;hf_xxx&quot;</span>}',wrap:!1}}),X=new as({props:{title:"Read",local:"read",headingTag:"h2"}}),Y=new o({props:{code:" ",highlighted:`<span class="hljs-keyword">from</span> functools <span class="hljs-keyword">import</span> partial
<span class="hljs-keyword">from</span> typing <span class="hljs-keyword">import</span> Iterator, <span class="hljs-type">Optional</span>, <span class="hljs-type">Union</span>

<span class="hljs-keyword">import</span> pyarrow <span class="hljs-keyword">as</span> pa
<span class="hljs-keyword">import</span> pyarrow.parquet <span class="hljs-keyword">as</span> pq
<span class="hljs-keyword">from</span> huggingface_hub <span class="hljs-keyword">import</span> HfFileSystem
<span class="hljs-keyword">from</span> pyspark.sql.dataframe <span class="hljs-keyword">import</span> DataFrame
<span class="hljs-keyword">from</span> pyspark.sql.pandas.types <span class="hljs-keyword">import</span> from_arrow_schema


<span class="hljs-keyword">def</span> <span class="hljs-title function_">_read</span>(<span class="hljs-params">iterator: Iterator[pa.RecordBatch], columns: <span class="hljs-type">Optional</span>[<span class="hljs-built_in">list</span>[<span class="hljs-built_in">str</span>]], filters: <span class="hljs-type">Optional</span>[<span class="hljs-type">Union</span>[<span class="hljs-built_in">list</span>[<span class="hljs-built_in">tuple</span>], <span class="hljs-built_in">list</span>[<span class="hljs-built_in">list</span>[<span class="hljs-built_in">tuple</span>]]]], **kwargs</span>) -&gt; Iterator[pa.RecordBatch]:
    <span class="hljs-keyword">for</span> batch <span class="hljs-keyword">in</span> iterator:
        paths = batch[<span class="hljs-number">0</span>].to_pylist()
        ds = pq.ParquetDataset(paths, **kwargs)
        <span class="hljs-keyword">yield</span> <span class="hljs-keyword">from</span> ds._dataset.to_batches(columns=columns, <span class="hljs-built_in">filter</span>=pq.filters_to_expression(filters) <span class="hljs-keyword">if</span> filters <span class="hljs-keyword">else</span> <span class="hljs-literal">None</span>)


<span class="hljs-keyword">def</span> <span class="hljs-title function_">read_parquet</span>(<span class="hljs-params">
    path: <span class="hljs-built_in">str</span>,
    columns: <span class="hljs-type">Optional</span>[<span class="hljs-built_in">list</span>[<span class="hljs-built_in">str</span>]] = <span class="hljs-literal">None</span>,
    filters: <span class="hljs-type">Optional</span>[<span class="hljs-type">Union</span>[<span class="hljs-built_in">list</span>[<span class="hljs-built_in">tuple</span>], <span class="hljs-built_in">list</span>[<span class="hljs-built_in">list</span>[<span class="hljs-built_in">tuple</span>]]]] = <span class="hljs-literal">None</span>,
    **kwargs,
</span>) -&gt; DataFrame:
    <span class="hljs-string">&quot;&quot;&quot;
    Loads Parquet files from Hugging Face using PyArrow, returning a PySPark \`DataFrame\`.

    It reads Parquet files in a distributed manner.

    Access private or gated repositories using \`huggingface-cli login\` or passing a token
    using the \`storage_options\` argument: \`storage_options={&quot;token&quot;: &quot;hf_xxx&quot;}\`

    Parameters
    ----------
    path : str
        Path to the file. Prefix with a protocol like \`hf://\` to read from Hugging Face.
        You can read from multiple files if you pass a globstring.
    columns : list, default None
        If not None, only these columns will be read from the file.
    filters : List[Tuple] or List[List[Tuple]], default None
        To filter out data.
        Filter syntax: [[(column, op, val), ...],...]
        where op is [==, =, &gt;, &gt;=, &lt;, &lt;=, !=, in, not in]
        The innermost tuples are transposed into a set of filters applied
        through an \`AND\` operation.
        The outer list combines these sets of filters through an \`OR\`
        operation.
        A single list of tuples can also be used, meaning that no \`OR\`
        operation between set of filters is to be conducted.

    **kwargs
        Any additional kwargs are passed to pyarrow.parquet.ParquetDataset.

    Returns
    -------
    DataFrame
        DataFrame based on parquet file.

    Examples
    --------
    &gt;&gt;&gt; path = &quot;hf://datasets/username/dataset/data.parquet&quot;
    &gt;&gt;&gt; pd.DataFrame({&quot;foo&quot;: range(5), &quot;bar&quot;: range(5, 10)}).to_parquet(path)
    &gt;&gt;&gt; read_parquet(path).show()
    +---+---+
    |foo|bar|
    +---+---+
    |  0|  5|
    |  1|  6|
    |  2|  7|
    |  3|  8|
    |  4|  9|
    +---+---+
    &gt;&gt;&gt; read_parquet(path, columns=[&quot;bar&quot;]).show()
    +---+
    |bar|
    +---+
    |  5|
    |  6|
    |  7|
    |  8|
    |  9|
    +---+
    &gt;&gt;&gt; sel = [(&quot;foo&quot;, &quot;&gt;&quot;, 2)]
    &gt;&gt;&gt; read_parquet(path, filters=sel).show()
    +---+---+
    |foo|bar|
    +---+---+
    |  3|  8|
    |  4|  9|
    +---+---+
    &quot;&quot;&quot;</span>
    filesystem: HfFileSystem = kwargs.pop(<span class="hljs-string">&quot;filesystem&quot;</span>) <span class="hljs-keyword">if</span> <span class="hljs-string">&quot;filesystem&quot;</span> <span class="hljs-keyword">in</span> kwargs <span class="hljs-keyword">else</span> HfFileSystem(**kwargs.pop(<span class="hljs-string">&quot;storage_options&quot;</span>, {}))
    paths = filesystem.glob(path)
    <span class="hljs-keyword">if</span> <span class="hljs-keyword">not</span> paths:
        <span class="hljs-keyword">raise</span> FileNotFoundError(<span class="hljs-string">f&quot;Counldn&#x27;t find any file at <span class="hljs-subst">{path}</span>&quot;</span>)
    rdd = spark.sparkContext.parallelize([{<span class="hljs-string">&quot;path&quot;</span>: path} <span class="hljs-keyword">for</span> path <span class="hljs-keyword">in</span> paths], <span class="hljs-built_in">len</span>(paths))
    df = spark.createDataFrame(rdd)
    arrow_schema = pq.read_schema(filesystem.<span class="hljs-built_in">open</span>(paths[<span class="hljs-number">0</span>]))
    schema = pa.schema([field <span class="hljs-keyword">for</span> field <span class="hljs-keyword">in</span> arrow_schema <span class="hljs-keyword">if</span> (columns <span class="hljs-keyword">is</span> <span class="hljs-literal">None</span> <span class="hljs-keyword">or</span> field.name <span class="hljs-keyword">in</span> columns)], metadata=arrow_schema.metadata)
    <span class="hljs-keyword">return</span> df.mapInArrow(
        partial(_read, columns=columns, filters=filters, filesystem=filesystem, schema=arrow_schema, **kwargs),
        from_arrow_schema(schema),
    )`,wrap:!1}}),E=new o({props:{code:"ZnJvbSUyMHB5c3Bhcmsuc3FsJTIwaW1wb3J0JTIwU3BhcmtTZXNzaW9uJTBBc3BhcmslMjAlM0QlMjBTcGFya1Nlc3Npb24uYnVpbGRlci5hcHBOYW1lKCUyMmRlbW8lMjIpLmdldE9yQ3JlYXRlKCklMEFkZiUyMCUzRCUyMHJlYWRfcGFycXVldCglMjJoZiUzQSUyRiUyRmRhdGFzZXRzJTJGQkFBSSUyRkluZmluaXR5LUluc3RydWN0JTJGN00lMkYqLnBhcnF1ZXQlMjIpJTBBZGYuc2hvdygp",highlighted:`<span class="hljs-meta">&gt;&gt;&gt; </span><span class="hljs-keyword">from</span> pyspark.sql <span class="hljs-keyword">import</span> SparkSession
<span class="hljs-meta">&gt;&gt;&gt; </span>spark = SparkSession.builder.appName(<span class="hljs-string">&quot;demo&quot;</span>).getOrCreate()
<span class="hljs-meta">&gt;&gt;&gt; </span>df = read_parquet(<span class="hljs-string">&quot;hf://datasets/BAAI/Infinity-Instruct/7M/*.parquet&quot;</span>)
<span class="hljs-meta">&gt;&gt;&gt; </span>df.show()
+---+----------------------------+-----+----------+--------------------+        
| <span class="hljs-built_in">id</span>|               conversations|label|langdetect|              source|
+---+----------------------------+-----+----------+--------------------+
|  <span class="hljs-number">0</span>|        [{human, <span class="hljs-keyword">def</span> <span class="hljs-title function_">exti</span>...|     |        en|      code_exercises|
|  <span class="hljs-number">1</span>|        [{human, See the ...|     |        en|                flan|
|  <span class="hljs-number">2</span>|        [{human, This <span class="hljs-keyword">is</span> ...|     |        en|                flan|
|  <span class="hljs-number">3</span>|        [{human, If you d...|     |        en|                flan|
|  <span class="hljs-number">4</span>|        [{human, In a Uni...|     |        en|                flan|
|  <span class="hljs-number">5</span>|        [{human, Read the...|     |        en|                flan|
|  <span class="hljs-number">6</span>|        [{human, You are ...|     |        en|          code_bagel|
|  <span class="hljs-number">7</span>|        [{human, I want y...|     |        en|          Subjective|
|  <span class="hljs-number">8</span>|        [{human, Given th...|     |        en|                flan|
|  <span class="hljs-number">9</span>|[{human, 因果联系原则是法...|     |     zh-cn|          Subjective|
| <span class="hljs-number">10</span>|        [{human, Provide ...|     |        en|self-oss-instruct...|
| <span class="hljs-number">11</span>|        [{human, The univ...|     |        en|                flan|
| <span class="hljs-number">12</span>|        [{human, Q: I am ...|     |        en|                flan|
| <span class="hljs-number">13</span>|        [{human, What <span class="hljs-keyword">is</span> ...|     |        en|      OpenHermes-<span class="hljs-number">2.5</span>|
| <span class="hljs-number">14</span>|        [{human, In react...|     |        en|                flan|
| <span class="hljs-number">15</span>|        [{human, Write Py...|     |        en|      code_exercises|
| <span class="hljs-number">16</span>|        [{human, Find the...|     |        en|            MetaMath|
| <span class="hljs-number">17</span>|        [{human, Three of...|     |        en|            MetaMath|
| <span class="hljs-number">18</span>|        [{human, Chandra ...|     |        en|            MetaMath|
| <span class="hljs-number">19</span>|[{human, 用经济学知识分析...|     |     zh-cn|          Subjective|
+---+----------------------------+-----+----------+--------------------+`,wrap:!1}}),z=new o({props:{code:"ZGZfbGFuZ2RldGVjdF9vbmx5JTIwJTNEJTIwcmVhZF9wYXJxdWV0KCUyMmhmJTNBJTJGJTJGZGF0YXNldHMlMkZCQUFJJTJGSW5maW5pdHktSW5zdHJ1Y3QlMkY3TSUyRioucGFycXVldCUyMiUyQyUyMGNvbHVtbnMlM0QlNUIlMjJsYW5nZGV0ZWN0JTIyJTVEKSUwQWRmX2xhbmdkZXRlY3Rfb25seS5ncm91cEJ5KCUyMmxhbmdkZXRlY3QlMjIpLmNvdW50KCkuc2hvdygp",highlighted:`<span class="hljs-meta">&gt;&gt;&gt; </span>df_langdetect_only = read_parquet(<span class="hljs-string">&quot;hf://datasets/BAAI/Infinity-Instruct/7M/*.parquet&quot;</span>, columns=[<span class="hljs-string">&quot;langdetect&quot;</span>])
<span class="hljs-meta">&gt;&gt;&gt; </span>df_langdetect_only.groupBy(<span class="hljs-string">&quot;langdetect&quot;</span>).count().show()
+----------+-------+                                                            
|langdetect|  count|
+----------+-------+
|        en|<span class="hljs-number">6697793</span>|
|     zh-cn| <span class="hljs-number">751313</span>|
+----------+-------+`,wrap:!1}}),q=new o({props:{code:"Y3JpdGVyaWElMjAlM0QlMjAlNUIoJTIybGFuZ2RldGVjdCUyMiUyQyUyMCUyMiUzRCUyMiUyQyUyMCUyMnpoLWNuJTIyKSU1RCUwQWRmX2NoaW5lc2Vfb25seSUyMCUzRCUyMHJlYWRfcGFycXVldCglMjJoZiUzQSUyRiUyRmRhdGFzZXRzJTJGQkFBSSUyRkluZmluaXR5LUluc3RydWN0JTJGN00lMkYqLnBhcnF1ZXQlMjIlMkMlMjBmaWx0ZXJzJTNEY3JpdGVyaWEpJTBBZGZfY2hpbmVzZV9vbmx5",highlighted:`<span class="hljs-meta">&gt;&gt;&gt; </span>criteria = [(<span class="hljs-string">&quot;langdetect&quot;</span>, <span class="hljs-string">&quot;=&quot;</span>, <span class="hljs-string">&quot;zh-cn&quot;</span>)]
<span class="hljs-meta">&gt;&gt;&gt; </span>df_chinese_only = read_parquet(<span class="hljs-string">&quot;hf://datasets/BAAI/Infinity-Instruct/7M/*.parquet&quot;</span>, filters=criteria)
<span class="hljs-meta">&gt;&gt;&gt; </span>df_chinese_only
+---+----------------------------+-----+----------+----------+                  
| <span class="hljs-built_in">id</span>|               conversations|label|langdetect|    source|
+---+----------------------------+-----+----------+----------+
|  <span class="hljs-number">9</span>|[{human, 因果联系原则是法...|     |     zh-cn|Subjective|
| <span class="hljs-number">19</span>|[{human, 用经济学知识分析...|     |     zh-cn|Subjective|
| <span class="hljs-number">38</span>| [{human, 某个考试共有A、...|     |     zh-cn|Subjective|
| <span class="hljs-number">39</span>|[{human, 撰写一篇关于斐波...|     |     zh-cn|Subjective|
| <span class="hljs-number">57</span>|[{human, 总结世界历史上的...|     |     zh-cn|Subjective|
| <span class="hljs-number">61</span>|[{human, 生成一则广告词。...|     |     zh-cn|Subjective|
| <span class="hljs-number">66</span>|[{human, 描述一个有效的团...|     |     zh-cn|Subjective|
| <span class="hljs-number">94</span>|[{human, 如果比利和蒂芙尼...|     |     zh-cn|Subjective|
|<span class="hljs-number">102</span>|[{human, 生成一句英文名言...|     |     zh-cn|Subjective|
|<span class="hljs-number">106</span>|[{human, 写一封感谢信，感...|     |     zh-cn|Subjective|
|<span class="hljs-number">118</span>| [{human, 生成一个故事。}...|     |     zh-cn|Subjective|
|<span class="hljs-number">174</span>|[{human, 高胆固醇水平的后...|     |     zh-cn|Subjective|
|<span class="hljs-number">180</span>|[{human, 基于以下角色信息...|     |     zh-cn|Subjective|
|<span class="hljs-number">192</span>|[{human, 请写一篇文章，概...|     |     zh-cn|Subjective|
|<span class="hljs-number">221</span>|[{human, 以诗歌形式表达对...|     |     zh-cn|Subjective|
|<span class="hljs-number">228</span>|[{human, 根据给定的指令，...|     |     zh-cn|Subjective|
|<span class="hljs-number">236</span>|[{human, 打开一个新的生成...|     |     zh-cn|Subjective|
|<span class="hljs-number">260</span>|[{human, 生成一个有关未来...|     |     zh-cn|Subjective|
|<span class="hljs-number">268</span>|[{human, 如果有一定数量的...|     |     zh-cn|Subjective|
|<span class="hljs-number">273</span>| [{human, 题目：小明有<span class="hljs-number">5</span>个...|     |     zh-cn|Subjective|
+---+----------------------------+-----+----------+----------+`,wrap:!1}}),H=new as({props:{title:"Write",local:"write",headingTag:"h2"}}),$=new o({props:{code:" ",highlighted:`<span class="hljs-keyword">import</span> math
<span class="hljs-keyword">import</span> pickle
<span class="hljs-keyword">import</span> tempfile
<span class="hljs-keyword">from</span> functools <span class="hljs-keyword">import</span> partial
<span class="hljs-keyword">from</span> typing <span class="hljs-keyword">import</span> Iterator, <span class="hljs-type">Optional</span>

<span class="hljs-keyword">import</span> pyarrow <span class="hljs-keyword">as</span> pa
<span class="hljs-keyword">import</span> pyarrow.parquet <span class="hljs-keyword">as</span> pq
<span class="hljs-keyword">from</span> huggingface_hub <span class="hljs-keyword">import</span> CommitOperationAdd, HfFileSystem
<span class="hljs-keyword">from</span> pyspark.sql.dataframe <span class="hljs-keyword">import</span> DataFrame
<span class="hljs-keyword">from</span> pyspark.sql.pandas.types <span class="hljs-keyword">import</span> from_arrow_schema, to_arrow_schema


<span class="hljs-keyword">def</span> <span class="hljs-title function_">_preupload</span>(<span class="hljs-params">iterator: Iterator[pa.RecordBatch], path: <span class="hljs-built_in">str</span>, schema: pa.Schema, filesystem: HfFileSystem, row_group_size: <span class="hljs-type">Optional</span>[<span class="hljs-built_in">int</span>] = <span class="hljs-literal">None</span>, **kwargs</span>) -&gt; Iterator[pa.RecordBatch]:
    resolved_path = filesystem.resolve_path(path)
    <span class="hljs-keyword">with</span> tempfile.NamedTemporaryFile(suffix=<span class="hljs-string">&quot;.parquet&quot;</span>) <span class="hljs-keyword">as</span> temp_file:
        <span class="hljs-keyword">with</span> pq.ParquetWriter(temp_file.name, schema=schema, **kwargs) <span class="hljs-keyword">as</span> writer:
            <span class="hljs-keyword">for</span> batch <span class="hljs-keyword">in</span> iterator:
                writer.write_batch(batch, row_group_size=row_group_size)
        addition = CommitOperationAdd(path_in_repo=temp_file.name, path_or_fileobj=temp_file.name)
        filesystem._api.preupload_lfs_files(repo_id=resolved_path.repo_id, additions=[addition], repo_type=resolved_path.repo_type, revision=resolved_path.revision)
    <span class="hljs-keyword">yield</span> pa.record_batch({<span class="hljs-string">&quot;addition&quot;</span>: [pickle.dumps(addition)]}, schema=pa.schema({<span class="hljs-string">&quot;addition&quot;</span>: pa.binary()}))


<span class="hljs-keyword">def</span> <span class="hljs-title function_">_commit</span>(<span class="hljs-params">iterator: Iterator[pa.RecordBatch], path: <span class="hljs-built_in">str</span>, filesystem: HfFileSystem, max_operations_per_commit=<span class="hljs-number">50</span></span>) -&gt; Iterator[pa.RecordBatch]:
    resolved_path = filesystem.resolve_path(path)
    additions: <span class="hljs-built_in">list</span>[CommitOperationAdd] = [pickle.loads(addition) <span class="hljs-keyword">for</span> addition <span class="hljs-keyword">in</span> pa.Table.from_batches(iterator, schema=pa.schema({<span class="hljs-string">&quot;addition&quot;</span>: pa.binary()}))[<span class="hljs-number">0</span>].to_pylist()]
    num_commits = math.ceil(<span class="hljs-built_in">len</span>(additions) / max_operations_per_commit)
    <span class="hljs-keyword">for</span> shard_idx, addition <span class="hljs-keyword">in</span> <span class="hljs-built_in">enumerate</span>(additions):
        addition.path_in_repo = resolved_path.path_in_repo.replace(<span class="hljs-string">&quot;{shard_idx:05d}&quot;</span>, <span class="hljs-string">f&quot;<span class="hljs-subst">{shard_idx:05d}</span>&quot;</span>)
    <span class="hljs-keyword">for</span> i <span class="hljs-keyword">in</span> <span class="hljs-built_in">range</span>(<span class="hljs-number">0</span>, num_commits):
        operations = additions[i * max_operations_per_commit : (i + <span class="hljs-number">1</span>) * max_operations_per_commit]
        commit_message = <span class="hljs-string">&quot;Upload using PySpark&quot;</span> + (<span class="hljs-string">f&quot; (part <span class="hljs-subst">{i:05d}</span>-of-<span class="hljs-subst">{num_commits:05d}</span>)&quot;</span> <span class="hljs-keyword">if</span> num_commits &gt; <span class="hljs-number">1</span> <span class="hljs-keyword">else</span> <span class="hljs-string">&quot;&quot;</span>)
        filesystem._api.create_commit(repo_id=resolved_path.repo_id, repo_type=resolved_path.repo_type, revision=resolved_path.revision, operations=operations, commit_message=commit_message)
        <span class="hljs-keyword">yield</span> pa.record_batch({<span class="hljs-string">&quot;path&quot;</span>: [addition.path_in_repo <span class="hljs-keyword">for</span> addition <span class="hljs-keyword">in</span> operations]}, schema=pa.schema({<span class="hljs-string">&quot;path&quot;</span>: pa.string()}))


<span class="hljs-keyword">def</span> <span class="hljs-title function_">write_parquet</span>(<span class="hljs-params">df: DataFrame, path: <span class="hljs-built_in">str</span>, **kwargs</span>) -&gt; <span class="hljs-literal">None</span>:
    <span class="hljs-string">&quot;&quot;&quot;
    Write Parquet files to Hugging Face using PyArrow.

    It uploads Parquet files in a distributed manner in two steps:

    1. Preupload the Parquet files in parallel in a distributed banner
    2. Commit the preuploaded files

    Authenticate using \`huggingface-cli login\` or passing a token
    using the \`storage_options\` argument: \`storage_options={&quot;token&quot;: &quot;hf_xxx&quot;}\`

    Parameters
    ----------
    path : str
        Path of the file or directory. Prefix with a protocol like \`hf://\` to read from Hugging Face.
        It writes Parquet files in the form &quot;part-xxxxx.parquet&quot;, or to a single file if \`path ends with &quot;.parquet&quot;.

    **kwargs
        Any additional kwargs are passed to pyarrow.parquet.ParquetWriter.

    Returns
    -------
    DataFrame
        DataFrame based on parquet file.

    Examples
    --------
    &gt;&gt;&gt; spark.createDataFrame(pd.DataFrame({&quot;foo&quot;: range(5), &quot;bar&quot;: range(5, 10)}))
    &gt;&gt;&gt; # Save to one file
    &gt;&gt;&gt; write_parquet(df, &quot;hf://datasets/username/dataset/data.parquet&quot;)
    &gt;&gt;&gt; # OR save to a directory (possibly in many files)
    &gt;&gt;&gt; write_parquet(df, &quot;hf://datasets/username/dataset&quot;)
    &quot;&quot;&quot;</span>
    filesystem: HfFileSystem = kwargs.pop(<span class="hljs-string">&quot;filesystem&quot;</span>, HfFileSystem(**kwargs.pop(<span class="hljs-string">&quot;storage_options&quot;</span>, {})))
    <span class="hljs-keyword">if</span> path.endswith(<span class="hljs-string">&quot;.parquet&quot;</span>) <span class="hljs-keyword">or</span> path.endswith(<span class="hljs-string">&quot;.pq&quot;</span>):
        df = df.coalesce(<span class="hljs-number">1</span>)
    <span class="hljs-keyword">else</span>:
        path += <span class="hljs-string">&quot;/part-{shard_idx:05d}.parquet&quot;</span>
    df.mapInArrow(
        partial(_preupload, path=path, schema=to_arrow_schema(df.schema), filesystem=filesystem, **kwargs),
        from_arrow_schema(pa.schema({<span class="hljs-string">&quot;addition&quot;</span>: pa.binary()})),
    ).repartition(<span class="hljs-number">1</span>).mapInArrow(
        partial(_commit, path=path, filesystem=filesystem),
        from_arrow_schema(pa.schema({<span class="hljs-string">&quot;path&quot;</span>: pa.string()})),
    ).collect()`,wrap:!1}}),P=new o({props:{code:"d3JpdGVfcGFycXVldChkZl9jaGluZXNlX29ubHklMkMlMjAlMjJoZiUzQSUyRiUyRmRhdGFzZXRzJTJGdXNlcm5hbWUlMkZJbmZpbml0eS1JbnN0cnVjdC1DaGluZXNlLU9ubHklMjIp",highlighted:`<span class="hljs-meta">&gt;&gt;&gt; </span>write_parquet(df_chinese_only, <span class="hljs-string">&quot;hf://datasets/username/Infinity-Instruct-Chinese-Only&quot;</span>)
tmph9jwu9py.parquet: <span class="hljs-number">100</span>%|██████████| <span class="hljs-number">50.5</span>M/<span class="hljs-number">50.5</span>M [<span class="hljs-number">00</span>:03&lt;<span class="hljs-number">00</span>:<span class="hljs-number">00</span>, <span class="hljs-number">14.6</span>MB/s]
tmp0oqt99nc.parquet: <span class="hljs-number">100</span>%|██████████| <span class="hljs-number">50.8</span>M/<span class="hljs-number">50.8</span>M [<span class="hljs-number">00</span>:02&lt;<span class="hljs-number">00</span>:<span class="hljs-number">00</span>, <span class="hljs-number">17.9</span>MB/s]
tmpgnizkwqp.parquet: <span class="hljs-number">100</span>%|██████████| <span class="hljs-number">50.5</span>M/<span class="hljs-number">50.5</span>M [<span class="hljs-number">00</span>:02&lt;<span class="hljs-number">00</span>:<span class="hljs-number">00</span>, <span class="hljs-number">19.6</span>MB/s]
tmpanm04k4n.parquet: <span class="hljs-number">100</span>%|██████████| <span class="hljs-number">51.4</span>M/<span class="hljs-number">51.4</span>M [<span class="hljs-number">00</span>:02&lt;<span class="hljs-number">00</span>:<span class="hljs-number">00</span>, <span class="hljs-number">22.9</span>MB/s]
tmp14uy9oqb.parquet: <span class="hljs-number">100</span>%|██████████| <span class="hljs-number">50.4</span>M/<span class="hljs-number">50.4</span>M [<span class="hljs-number">00</span>:02&lt;<span class="hljs-number">00</span>:<span class="hljs-number">00</span>, <span class="hljs-number">23.0</span>MB/s]
tmpcp8t_qdl.parquet: <span class="hljs-number">100</span>%|██████████| <span class="hljs-number">50.4</span>M/<span class="hljs-number">50.4</span>M [<span class="hljs-number">00</span>:02&lt;<span class="hljs-number">00</span>:<span class="hljs-number">00</span>, <span class="hljs-number">23.5</span>MB/s]
tmpjui5mns8.parquet: <span class="hljs-number">100</span>%|██████████| <span class="hljs-number">50.3</span>M/<span class="hljs-number">50.3</span>M [<span class="hljs-number">00</span>:02&lt;<span class="hljs-number">00</span>:<span class="hljs-number">00</span>, <span class="hljs-number">24.1</span>MB/s]
tmpydqh6od1.parquet: <span class="hljs-number">100</span>%|██████████| <span class="hljs-number">50.9</span>M/<span class="hljs-number">50.9</span>M [<span class="hljs-number">00</span>:02&lt;<span class="hljs-number">00</span>:<span class="hljs-number">00</span>, <span class="hljs-number">23.8</span>MB/s]
tmp52f2t8tu.parquet: <span class="hljs-number">100</span>%|██████████| <span class="hljs-number">50.5</span>M/<span class="hljs-number">50.5</span>M [<span class="hljs-number">00</span>:02&lt;<span class="hljs-number">00</span>:<span class="hljs-number">00</span>, <span class="hljs-number">23.7</span>MB/s]
tmpg7egv3ye.parquet: <span class="hljs-number">100</span>%|██████████| <span class="hljs-number">50.1</span>M/<span class="hljs-number">50.1</span>M [<span class="hljs-number">00</span>:06&lt;<span class="hljs-number">00</span>:<span class="hljs-number">00</span>, <span class="hljs-number">7.68</span>MB/s]
tmp2s0fq2hm.parquet: <span class="hljs-number">100</span>%|██████████| <span class="hljs-number">50.8</span>M/<span class="hljs-number">50.8</span>M [<span class="hljs-number">00</span>:02&lt;<span class="hljs-number">00</span>:<span class="hljs-number">00</span>, <span class="hljs-number">18.1</span>MB/s]
tmpmj97ab30.parquet: <span class="hljs-number">100</span>%|██████████| <span class="hljs-number">71.3</span>M/<span class="hljs-number">71.3</span>M [<span class="hljs-number">00</span>:02&lt;<span class="hljs-number">00</span>:<span class="hljs-number">00</span>, <span class="hljs-number">23.9</span>MB/s]`,wrap:!1}}),O=new Tl({props:{source:"https://github.com/huggingface/hub-docs/blob/main/docs/hub/datasets-spark.md"}}),{c(){j=p("meta"),ns=t(),ss=p("p"),ts=t(),c(u.$$.fragment),es=t(),T=p("p"),T.textContent=qs,ps=t(),d=p("p"),d.innerHTML=Hs,Ms=t(),c(b.$$.fragment),ys=t(),I=p("p"),I.innerHTML=xs,cs=t(),c(C.$$.fragment),is=t(),B=p("p"),B.innerHTML=Ls,Js=t(),c(G.$$.fragment),rs=t(),c(f.$$.fragment),Us=t(),g=p("p"),g.textContent=$s,hs=t(),Z=p("p"),Z.textContent=Ds,os=t(),c(k.$$.fragment),js=t(),S=p("p"),S.innerHTML=Ks,ms=t(),c(R.$$.fragment),ws=t(),W=p("p"),W.innerHTML=Ps,us=t(),c(X.$$.fragment),Ts=t(),A=p("p"),A.textContent=Os,ds=t(),Q=p("p"),Q.innerHTML=sl,bs=t(),c(Y.$$.fragment),Is=t(),N=p("p"),N.innerHTML=ll,Cs=t(),m=p("div"),m.innerHTML=al,Bs=t(),F=p("p"),F.innerHTML=nl,Gs=t(),V=p("p"),V.textContent=tl,fs=t(),c(E.$$.fragment),gs=t(),v=p("p"),v.innerHTML=el,Zs=t(),c(z.$$.fragment),ks=t(),_=p("p"),_.textContent=pl,Ss=t(),c(q.$$.fragment),Rs=t(),c(H.$$.fragment),Ws=t(),x=p("p"),x.textContent=Ml,Xs=t(),L=p("p"),L.innerHTML=yl,As=t(),c($.$$.fragment),Qs=t(),D=p("p"),D.innerHTML=cl,Ys=t(),K=p("p"),K.innerHTML=il,Ns=t(),c(P.$$.fragment),Fs=t(),w=p("div"),w.innerHTML=Jl,Vs=t(),c(O.$$.fragment),Es=t(),ls=p("p"),this.h()},l(s){const l=wl("svelte-u9bgzb",document.head);j=M(l,"META",{name:!0,content:!0}),l.forEach(a),ns=e(s),ss=M(s,"P",{}),rl(ss).forEach(a),ts=e(s),i(u.$$.fragment,s),es=e(s),T=M(s,"P",{"data-svelte-h":!0}),y(T)!=="svelte-s0r53o"&&(T.textContent=qs),ps=e(s),d=M(s,"P",{"data-svelte-h":!0}),y(d)!=="svelte-12z8vci"&&(d.innerHTML=Hs),Ms=e(s),i(b.$$.fragment,s),ys=e(s),I=M(s,"P",{"data-svelte-h":!0}),y(I)!=="svelte-13h0lcm"&&(I.innerHTML=xs),cs=e(s),i(C.$$.fragment,s),is=e(s),B=M(s,"P",{"data-svelte-h":!0}),y(B)!=="svelte-ml7hwl"&&(B.innerHTML=Ls),Js=e(s),i(G.$$.fragment,s),rs=e(s),i(f.$$.fragment,s),Us=e(s),g=M(s,"P",{"data-svelte-h":!0}),y(g)!=="svelte-aqw7kt"&&(g.textContent=$s),hs=e(s),Z=M(s,"P",{"data-svelte-h":!0}),y(Z)!=="svelte-pdivv8"&&(Z.textContent=Ds),os=e(s),i(k.$$.fragment,s),js=e(s),S=M(s,"P",{"data-svelte-h":!0}),y(S)!=="svelte-1pzllua"&&(S.innerHTML=Ks),ms=e(s),i(R.$$.fragment,s),ws=e(s),W=M(s,"P",{"data-svelte-h":!0}),y(W)!=="svelte-ygg3x4"&&(W.innerHTML=Ps),us=e(s),i(X.$$.fragment,s),Ts=e(s),A=M(s,"P",{"data-svelte-h":!0}),y(A)!=="svelte-7gpw7a"&&(A.textContent=Os),ds=e(s),Q=M(s,"P",{"data-svelte-h":!0}),y(Q)!=="svelte-4fzrqm"&&(Q.innerHTML=sl),bs=e(s),i(Y.$$.fragment,s),Is=e(s),N=M(s,"P",{"data-svelte-h":!0}),y(N)!=="svelte-1vx9tic"&&(N.innerHTML=ll),Cs=e(s),m=M(s,"DIV",{class:!0,"data-svelte-h":!0}),y(m)!=="svelte-1ct11n9"&&(m.innerHTML=al),Bs=e(s),F=M(s,"P",{"data-svelte-h":!0}),y(F)!=="svelte-1p4gwvr"&&(F.innerHTML=nl),Gs=e(s),V=M(s,"P",{"data-svelte-h":!0}),y(V)!=="svelte-19d7o0j"&&(V.textContent=tl),fs=e(s),i(E.$$.fragment,s),gs=e(s),v=M(s,"P",{"data-svelte-h":!0}),y(v)!=="svelte-q1u30o"&&(v.innerHTML=el),Zs=e(s),i(z.$$.fragment,s),ks=e(s),_=M(s,"P",{"data-svelte-h":!0}),y(_)!=="svelte-e2gcj5"&&(_.textContent=pl),Ss=e(s),i(q.$$.fragment,s),Rs=e(s),i(H.$$.fragment,s),Ws=e(s),x=M(s,"P",{"data-svelte-h":!0}),y(x)!=="svelte-nrb4bc"&&(x.textContent=Ml),Xs=e(s),L=M(s,"P",{"data-svelte-h":!0}),y(L)!=="svelte-104rxy3"&&(L.innerHTML=yl),As=e(s),i($.$$.fragment,s),Qs=e(s),D=M(s,"P",{"data-svelte-h":!0}),y(D)!=="svelte-1d35p3o"&&(D.innerHTML=cl),Ys=e(s),K=M(s,"P",{"data-svelte-h":!0}),y(K)!=="svelte-wkzo7"&&(K.innerHTML=il),Ns=e(s),i(P.$$.fragment,s),Fs=e(s),w=M(s,"DIV",{class:!0,"data-svelte-h":!0}),y(w)!=="svelte-tmflad"&&(w.innerHTML=Jl),Vs=e(s),i(O.$$.fragment,s),Es=e(s),ls=M(s,"P",{}),rl(ls).forEach(a),this.h()},h(){zs(j,"name","hf:doc:metadata"),zs(j,"content",bl),zs(m,"class","flex justify-center"),zs(w,"class","flex justify-center")},m(s,l){ul(document.head,j),n(s,ns,l),n(s,ss,l),n(s,ts,l),J(u,s,l),n(s,es,l),n(s,T,l),n(s,ps,l),n(s,d,l),n(s,Ms,l),J(b,s,l),n(s,ys,l),n(s,I,l),n(s,cs,l),J(C,s,l),n(s,is,l),n(s,B,l),n(s,Js,l),J(G,s,l),n(s,rs,l),J(f,s,l),n(s,Us,l),n(s,g,l),n(s,hs,l),n(s,Z,l),n(s,os,l),J(k,s,l),n(s,js,l),n(s,S,l),n(s,ms,l),J(R,s,l),n(s,ws,l),n(s,W,l),n(s,us,l),J(X,s,l),n(s,Ts,l),n(s,A,l),n(s,ds,l),n(s,Q,l),n(s,bs,l),J(Y,s,l),n(s,Is,l),n(s,N,l),n(s,Cs,l),n(s,m,l),n(s,Bs,l),n(s,F,l),n(s,Gs,l),n(s,V,l),n(s,fs,l),J(E,s,l),n(s,gs,l),n(s,v,l),n(s,Zs,l),J(z,s,l),n(s,ks,l),n(s,_,l),n(s,Ss,l),J(q,s,l),n(s,Rs,l),J(H,s,l),n(s,Ws,l),n(s,x,l),n(s,Xs,l),n(s,L,l),n(s,As,l),J($,s,l),n(s,Qs,l),n(s,D,l),n(s,Ys,l),n(s,K,l),n(s,Ns,l),J(P,s,l),n(s,Fs,l),n(s,w,l),n(s,Vs,l),J(O,s,l),n(s,Es,l),n(s,ls,l),vs=!0},p:hl,i(s){vs||(r(u.$$.fragment,s),r(b.$$.fragment,s),r(C.$$.fragment,s),r(G.$$.fragment,s),r(f.$$.fragment,s),r(k.$$.fragment,s),r(R.$$.fragment,s),r(X.$$.fragment,s),r(Y.$$.fragment,s),r(E.$$.fragment,s),r(z.$$.fragment,s),r(q.$$.fragment,s),r(H.$$.fragment,s),r($.$$.fragment,s),r(P.$$.fragment,s),r(O.$$.fragment,s),vs=!0)},o(s){U(u.$$.fragment,s),U(b.$$.fragment,s),U(C.$$.fragment,s),U(G.$$.fragment,s),U(f.$$.fragment,s),U(k.$$.fragment,s),U(R.$$.fragment,s),U(X.$$.fragment,s),U(Y.$$.fragment,s),U(E.$$.fragment,s),U(z.$$.fragment,s),U(q.$$.fragment,s),U(H.$$.fragment,s),U($.$$.fragment,s),U(P.$$.fragment,s),U(O.$$.fragment,s),vs=!1},d(s){s&&(a(ns),a(ss),a(ts),a(es),a(T),a(ps),a(d),a(Ms),a(ys),a(I),a(cs),a(is),a(B),a(Js),a(rs),a(Us),a(g),a(hs),a(Z),a(os),a(js),a(S),a(ms),a(ws),a(W),a(us),a(Ts),a(A),a(ds),a(Q),a(bs),a(Is),a(N),a(Cs),a(m),a(Bs),a(F),a(Gs),a(V),a(fs),a(gs),a(v),a(Zs),a(ks),a(_),a(Ss),a(Rs),a(Ws),a(x),a(Xs),a(L),a(As),a(Qs),a(D),a(Ys),a(K),a(Ns),a(Fs),a(w),a(Vs),a(Es),a(ls)),a(j),h(u,s),h(b,s),h(C,s),h(G,s),h(f,s),h(k,s),h(R,s),h(X,s),h(Y,s),h(E,s),h(z,s),h(q,s),h(H,s),h($,s),h(P,s),h(O,s)}}}const bl='{"title":"Spark","local":"spark","sections":[{"title":"Installation","local":"installation","sections":[],"depth":2},{"title":"Authentication","local":"authentication","sections":[],"depth":2},{"title":"Read","local":"read","sections":[],"depth":2},{"title":"Write","local":"write","sections":[],"depth":2}],"depth":1}';function Il(_s){return ol(()=>{new URLSearchParams(window.location.search).get("fw")}),[]}class gl extends jl{constructor(j){super(),ml(this,j,Il,dl,Ul,{})}}export{gl as component};
