import{s as Cl,y as Bl,n as fl,o as Gl}from"../chunks/scheduler.d6170356.js";import{S as gl,i as Zl,g as p,s as e,r as c,A as kl,h as M,f as a,c as n,j as Il,u as i,x as y,k as Ms,y as Sl,a as t,v as r,d as J,t as h,w as U}from"../chunks/index.fcd4cc08.js";import{C as o}from"../chunks/CodeBlock.7b16bdef.js";import{H as es,E as Rl}from"../chunks/EditOnGithub.da2b595c.js";function Wl(Ks){let m,ys,ns,cs,u,is,T,Os="Spark enables real-time, large-scale data processing in a distributed environment.",rs,d,sl="In particular you can use <code>huggingface_hub</code> to access Hugging Face datasets repositories in PySpark",Js,b,hs,I,ll="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:",Us,C,os,B,al="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>:",ms,f,js,G,ws,g,tl="You need to authenticate to Hugging Face to read private/gated dataset repositories or to write to your dataset repositories.",us,Z,el="You can use the CLI for example:",Ts,k,ds,S,nl="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:",bs,R,Is,W,pl='For more details about authentication, check out <a href="https://huggingface.co/docs/huggingface_hub/quick-start#authentication" rel="nofollow">this guide</a>.',Cs,X,Bs,Q,Ml="PySpark doesn’t have an official support for Hugging Face paths, so we provide a helper function to read datasets in a distributed manner.",fs,A,yl="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:",Gs,N,gs,Y,cl=`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.`,Zs,j,il='<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"/>',ks,F,rl="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.",Ss,V,Jl="After logging-in to access the gated repository, we can run:",Rs,v,Ws,E,hl=`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.`,Xs,z,Qs,_,Ul="To filter the dataset and only keep dialogues in Chinese:",As,q,Ns,H,Ys,x,ol="We also provide a helper function to write datasets in a distributed manner to a Hugging Face repository.",Fs,L,ml=`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:`,Vs,$,vs,D,jl='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.',Es,P,wl=`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:`,zs,K,_s,w,ul='<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"/>',qs,O,Hs,ss,Tl='You can duplicate the <a href="https://huggingface.co/spaces/lhoestq/Spark-on-HF-JupyterLab" rel="nofollow">Spark on HF JupyterLab</a> Space to get a Notebook with PySpark and those helper functions pre-installed.',xs,ls,dl="Click on “Duplicate Space”, choose a name for your Space, select your hardware and you are ready:",Ls,ts,bl,$s,as,Ds,ps,Ps;return u=new es({props:{title:"Spark",local:"spark",headingTag:"h1"}}),b=new es({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}}),f=new o({props:{code:"cGlwJTIwaW5zdGFsbCUyMHB5YXJyb3c=",highlighted:'pip <span class="hljs-keyword">install</span> pyarrow',wrap:!1}}),G=new es({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 es({props:{title:"Read",local:"read",headingTag:"h2"}}),N=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}}),v=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:"Y3JpdGVyaWElMjAlM0QlMjAlNUIoJTIybGFuZ2RldGVjdCUyMiUyQyUyMCUyMiUzRCUyMiUyQyUyMCUyMnpoLWNuJTIyKSU1RCUwQWRmX2NoaW5lc2Vfb25seSUyMCUzRCUyMHJlYWRfcGFycXVldCglMjJoZiUzQSUyRiUyRmRhdGFzZXRzJTJGQkFBSSUyRkluZmluaXR5LUluc3RydWN0JTJGN00lMkYqLnBhcnF1ZXQlMjIlMkMlMjBmaWx0ZXJzJTNEY3JpdGVyaWEpJTBBZGZfY2hpbmVzZV9vbmx5LnNob3coKQ==",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.show()
+---+----------------------------+-----+----------+----------+                  
| <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 es({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}}),K=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 es({props:{title:"Run in JupyterLab on Hugging Face Spaces",local:"run-in-jupyterlab-on-hugging-face-spaces",headingTag:"h2"}}),as=new Rl({props:{source:"https://github.com/huggingface/hub-docs/blob/main/docs/hub/datasets-spark.md"}}),{c(){m=p("meta"),ys=e(),ns=p("p"),cs=e(),c(u.$$.fragment),is=e(),T=p("p"),T.textContent=Os,rs=e(),d=p("p"),d.innerHTML=sl,Js=e(),c(b.$$.fragment),hs=e(),I=p("p"),I.innerHTML=ll,Us=e(),c(C.$$.fragment),os=e(),B=p("p"),B.innerHTML=al,ms=e(),c(f.$$.fragment),js=e(),c(G.$$.fragment),ws=e(),g=p("p"),g.textContent=tl,us=e(),Z=p("p"),Z.textContent=el,Ts=e(),c(k.$$.fragment),ds=e(),S=p("p"),S.innerHTML=nl,bs=e(),c(R.$$.fragment),Is=e(),W=p("p"),W.innerHTML=pl,Cs=e(),c(X.$$.fragment),Bs=e(),Q=p("p"),Q.textContent=Ml,fs=e(),A=p("p"),A.innerHTML=yl,Gs=e(),c(N.$$.fragment),gs=e(),Y=p("p"),Y.innerHTML=cl,Zs=e(),j=p("div"),j.innerHTML=il,ks=e(),F=p("p"),F.innerHTML=rl,Ss=e(),V=p("p"),V.textContent=Jl,Rs=e(),c(v.$$.fragment),Ws=e(),E=p("p"),E.innerHTML=hl,Xs=e(),c(z.$$.fragment),Qs=e(),_=p("p"),_.textContent=Ul,As=e(),c(q.$$.fragment),Ns=e(),c(H.$$.fragment),Ys=e(),x=p("p"),x.textContent=ol,Fs=e(),L=p("p"),L.innerHTML=ml,Vs=e(),c($.$$.fragment),vs=e(),D=p("p"),D.innerHTML=jl,Es=e(),P=p("p"),P.innerHTML=wl,zs=e(),c(K.$$.fragment),_s=e(),w=p("div"),w.innerHTML=ul,qs=e(),c(O.$$.fragment),Hs=e(),ss=p("p"),ss.innerHTML=Tl,xs=e(),ls=p("p"),ls.textContent=dl,Ls=e(),ts=p("img"),$s=e(),c(as.$$.fragment),Ds=e(),ps=p("p"),this.h()},l(s){const l=kl("svelte-u9bgzb",document.head);m=M(l,"META",{name:!0,content:!0}),l.forEach(a),ys=n(s),ns=M(s,"P",{}),Il(ns).forEach(a),cs=n(s),i(u.$$.fragment,s),is=n(s),T=M(s,"P",{"data-svelte-h":!0}),y(T)!=="svelte-s0r53o"&&(T.textContent=Os),rs=n(s),d=M(s,"P",{"data-svelte-h":!0}),y(d)!=="svelte-12z8vci"&&(d.innerHTML=sl),Js=n(s),i(b.$$.fragment,s),hs=n(s),I=M(s,"P",{"data-svelte-h":!0}),y(I)!=="svelte-13h0lcm"&&(I.innerHTML=ll),Us=n(s),i(C.$$.fragment,s),os=n(s),B=M(s,"P",{"data-svelte-h":!0}),y(B)!=="svelte-ml7hwl"&&(B.innerHTML=al),ms=n(s),i(f.$$.fragment,s),js=n(s),i(G.$$.fragment,s),ws=n(s),g=M(s,"P",{"data-svelte-h":!0}),y(g)!=="svelte-aqw7kt"&&(g.textContent=tl),us=n(s),Z=M(s,"P",{"data-svelte-h":!0}),y(Z)!=="svelte-pdivv8"&&(Z.textContent=el),Ts=n(s),i(k.$$.fragment,s),ds=n(s),S=M(s,"P",{"data-svelte-h":!0}),y(S)!=="svelte-1pzllua"&&(S.innerHTML=nl),bs=n(s),i(R.$$.fragment,s),Is=n(s),W=M(s,"P",{"data-svelte-h":!0}),y(W)!=="svelte-ygg3x4"&&(W.innerHTML=pl),Cs=n(s),i(X.$$.fragment,s),Bs=n(s),Q=M(s,"P",{"data-svelte-h":!0}),y(Q)!=="svelte-7gpw7a"&&(Q.textContent=Ml),fs=n(s),A=M(s,"P",{"data-svelte-h":!0}),y(A)!=="svelte-4fzrqm"&&(A.innerHTML=yl),Gs=n(s),i(N.$$.fragment,s),gs=n(s),Y=M(s,"P",{"data-svelte-h":!0}),y(Y)!=="svelte-1vx9tic"&&(Y.innerHTML=cl),Zs=n(s),j=M(s,"DIV",{class:!0,"data-svelte-h":!0}),y(j)!=="svelte-1ct11n9"&&(j.innerHTML=il),ks=n(s),F=M(s,"P",{"data-svelte-h":!0}),y(F)!=="svelte-1p4gwvr"&&(F.innerHTML=rl),Ss=n(s),V=M(s,"P",{"data-svelte-h":!0}),y(V)!=="svelte-19d7o0j"&&(V.textContent=Jl),Rs=n(s),i(v.$$.fragment,s),Ws=n(s),E=M(s,"P",{"data-svelte-h":!0}),y(E)!=="svelte-q1u30o"&&(E.innerHTML=hl),Xs=n(s),i(z.$$.fragment,s),Qs=n(s),_=M(s,"P",{"data-svelte-h":!0}),y(_)!=="svelte-e2gcj5"&&(_.textContent=Ul),As=n(s),i(q.$$.fragment,s),Ns=n(s),i(H.$$.fragment,s),Ys=n(s),x=M(s,"P",{"data-svelte-h":!0}),y(x)!=="svelte-nrb4bc"&&(x.textContent=ol),Fs=n(s),L=M(s,"P",{"data-svelte-h":!0}),y(L)!=="svelte-104rxy3"&&(L.innerHTML=ml),Vs=n(s),i($.$$.fragment,s),vs=n(s),D=M(s,"P",{"data-svelte-h":!0}),y(D)!=="svelte-1d35p3o"&&(D.innerHTML=jl),Es=n(s),P=M(s,"P",{"data-svelte-h":!0}),y(P)!=="svelte-wkzo7"&&(P.innerHTML=wl),zs=n(s),i(K.$$.fragment,s),_s=n(s),w=M(s,"DIV",{class:!0,"data-svelte-h":!0}),y(w)!=="svelte-tmflad"&&(w.innerHTML=ul),qs=n(s),i(O.$$.fragment,s),Hs=n(s),ss=M(s,"P",{"data-svelte-h":!0}),y(ss)!=="svelte-172t2p0"&&(ss.innerHTML=Tl),xs=n(s),ls=M(s,"P",{"data-svelte-h":!0}),y(ls)!=="svelte-1rabvow"&&(ls.textContent=dl),Ls=n(s),ts=M(s,"IMG",{src:!0}),$s=n(s),i(as.$$.fragment,s),Ds=n(s),ps=M(s,"P",{}),Il(ps).forEach(a),this.h()},h(){Ms(m,"name","hf:doc:metadata"),Ms(m,"content",Xl),Ms(j,"class","flex justify-center"),Ms(w,"class","flex justify-center"),Bl(ts.src,bl="https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/hub/spark-on-hf-jupyterlab-screenshot-min.png")||Ms(ts,"src",bl)},m(s,l){Sl(document.head,m),t(s,ys,l),t(s,ns,l),t(s,cs,l),r(u,s,l),t(s,is,l),t(s,T,l),t(s,rs,l),t(s,d,l),t(s,Js,l),r(b,s,l),t(s,hs,l),t(s,I,l),t(s,Us,l),r(C,s,l),t(s,os,l),t(s,B,l),t(s,ms,l),r(f,s,l),t(s,js,l),r(G,s,l),t(s,ws,l),t(s,g,l),t(s,us,l),t(s,Z,l),t(s,Ts,l),r(k,s,l),t(s,ds,l),t(s,S,l),t(s,bs,l),r(R,s,l),t(s,Is,l),t(s,W,l),t(s,Cs,l),r(X,s,l),t(s,Bs,l),t(s,Q,l),t(s,fs,l),t(s,A,l),t(s,Gs,l),r(N,s,l),t(s,gs,l),t(s,Y,l),t(s,Zs,l),t(s,j,l),t(s,ks,l),t(s,F,l),t(s,Ss,l),t(s,V,l),t(s,Rs,l),r(v,s,l),t(s,Ws,l),t(s,E,l),t(s,Xs,l),r(z,s,l),t(s,Qs,l),t(s,_,l),t(s,As,l),r(q,s,l),t(s,Ns,l),r(H,s,l),t(s,Ys,l),t(s,x,l),t(s,Fs,l),t(s,L,l),t(s,Vs,l),r($,s,l),t(s,vs,l),t(s,D,l),t(s,Es,l),t(s,P,l),t(s,zs,l),r(K,s,l),t(s,_s,l),t(s,w,l),t(s,qs,l),r(O,s,l),t(s,Hs,l),t(s,ss,l),t(s,xs,l),t(s,ls,l),t(s,Ls,l),t(s,ts,l),t(s,$s,l),r(as,s,l),t(s,Ds,l),t(s,ps,l),Ps=!0},p:fl,i(s){Ps||(J(u.$$.fragment,s),J(b.$$.fragment,s),J(C.$$.fragment,s),J(f.$$.fragment,s),J(G.$$.fragment,s),J(k.$$.fragment,s),J(R.$$.fragment,s),J(X.$$.fragment,s),J(N.$$.fragment,s),J(v.$$.fragment,s),J(z.$$.fragment,s),J(q.$$.fragment,s),J(H.$$.fragment,s),J($.$$.fragment,s),J(K.$$.fragment,s),J(O.$$.fragment,s),J(as.$$.fragment,s),Ps=!0)},o(s){h(u.$$.fragment,s),h(b.$$.fragment,s),h(C.$$.fragment,s),h(f.$$.fragment,s),h(G.$$.fragment,s),h(k.$$.fragment,s),h(R.$$.fragment,s),h(X.$$.fragment,s),h(N.$$.fragment,s),h(v.$$.fragment,s),h(z.$$.fragment,s),h(q.$$.fragment,s),h(H.$$.fragment,s),h($.$$.fragment,s),h(K.$$.fragment,s),h(O.$$.fragment,s),h(as.$$.fragment,s),Ps=!1},d(s){s&&(a(ys),a(ns),a(cs),a(is),a(T),a(rs),a(d),a(Js),a(hs),a(I),a(Us),a(os),a(B),a(ms),a(js),a(ws),a(g),a(us),a(Z),a(Ts),a(ds),a(S),a(bs),a(Is),a(W),a(Cs),a(Bs),a(Q),a(fs),a(A),a(Gs),a(gs),a(Y),a(Zs),a(j),a(ks),a(F),a(Ss),a(V),a(Rs),a(Ws),a(E),a(Xs),a(Qs),a(_),a(As),a(Ns),a(Ys),a(x),a(Fs),a(L),a(Vs),a(vs),a(D),a(Es),a(P),a(zs),a(_s),a(w),a(qs),a(Hs),a(ss),a(xs),a(ls),a(Ls),a(ts),a($s),a(Ds),a(ps)),a(m),U(u,s),U(b,s),U(C,s),U(f,s),U(G,s),U(k,s),U(R,s),U(X,s),U(N,s),U(v,s),U(z,s),U(q,s),U(H,s),U($,s),U(K,s),U(O,s),U(as,s)}}}const Xl='{"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},{"title":"Run in JupyterLab on Hugging Face Spaces","local":"run-in-jupyterlab-on-hugging-face-spaces","sections":[],"depth":2}],"depth":1}';function Ql(Ks){return Gl(()=>{new URLSearchParams(window.location.search).get("fw")}),[]}class Vl extends gl{constructor(m){super(),Zl(this,m,Ql,Wl,Cl,{})}}export{Vl as component};
