code: def from_dask_array(x, columns=None, index=None, meta=None):
      """Create a Dask DataFrame from a Dask Array.
  
      Converts a 2d array into a DataFrame and a 1d array into a Series.
  
      Parameters
      ----------
      x : da.Array
      columns : list or string
          list of column names if DataFrame, single string if Series
      index : dask.dataframe.Index, optional
          An optional *dask* Index to use for the output Series or DataFrame.
  
          The default output index depends on whether `x` has any unknown
          chunks. If there are any unknown chunks, the output has ``None``
          for all the divisions (one per chunk). If all the chunks are known,
          a default index with known divisions is created.
  
          Specifying `index` can be useful if you're conforming a Dask Array
          to an existing dask Series or DataFrame, and you would like the
          indices to match.
      meta : object, optional
          An optional `meta` parameter can be passed for dask
          to specify the concrete dataframe type to be returned.
          By default, pandas DataFrame is used.
  
      Examples
      --------
      >>> import dask.array as da
      >>> import dask.dataframe as dd
      >>> x = da.ones((4, 2), chunks=(2, 2))
      >>> df = dd.io.from_dask_array(x, columns=['a', 'b'])
      >>> df.compute()
           a    b
      0  1.0  1.0
      1  1.0  1.0
      2  1.0  1.0
      3  1.0  1.0
  
      See Also
      --------
      dask.bag.to_dataframe: from dask.bag
      dask.dataframe._Frame.values: Reverse conversion
      dask.dataframe._Frame.to_records: Reverse conversion
      """
      meta = _meta_from_array(x, columns, index, meta=meta)
  
      name = "from-dask-array-" + tokenize(x, columns)
      graph_dependencies = [x]
      arrays_and_indices = [x.name, "ij" if x.ndim == 2 else "i"]
      numblocks = {x.name: x.numblocks}
  
      if index is not None:
          # An index is explicitly given by the caller, so we can pass it through to the
          # initializer after a few checks.
          if index.npartitions != x.numblocks[0]:
              msg = (
                  "The index and array have different numbers of blocks. "
                  "({} != {})".format(index.npartitions, x.numblocks[0])
              )
              raise ValueError(msg)
          divisions = index.divisions
          graph_dependencies.append(index)
          arrays_and_indices.extend([index._name, "i"])
          numblocks[index._name] = (index.npartitions,)
  
      elif np.isnan(sum(x.shape)):
          # The shape of the incoming array is not known in at least one dimension. As
          # such, we can't create an index for the entire output DataFrame and we set
          # the divisions to None to represent that.
          divisions = [None] * (len(x.chunks[0]) + 1)
      else:
          # The shape of the incoming array is known and we don't have an explicit index.
          # Create a mapping of chunk number in the incoming array to
          # (start row, stop row) tuples. These tuples will be used to create a sequential
          # RangeIndex later on that is continuous over the whole DataFrame.
          n_elements = sum(x.chunks[0])
          divisions = [0]
          stop = 0
          index_mapping = {}
          for i, increment in enumerate(x.chunks[0]):
              stop += increment
              index_mapping[(i,)] = (divisions[i], stop)
  
              # last division corrected, even if there are empty chunk(s) at the end
              if stop == n_elements:
                  stop -= 1
  
              divisions.append(stop)
  
          arrays_and_indices.extend(
              [BlockwiseDepDict(mapping=index_mapping), "i"]
          )
  
      if is_series_like(meta):
          kwargs = {
              "dtype": x.dtype,
              "name": meta.name,
              "initializer": type(meta),
          }
      else:
          kwargs = {"columns": meta.columns, "initializer": type(meta)}
  
      blk = blockwise(
          _partition_from_array,
          name,
          "i",
          *arrays_and_indices,
          numblocks=numblocks,
          concatenate=True,
          # kwargs passed through to the DataFrame/Series initializer
          **kwargs,
      )
  
      graph = HighLevelGraph.from_collections(
          name, blk, dependencies=graph_dependencies
      )
      return new_dd_object(graph, name, meta, divisions)
  
imports: from __future__ import annotations
  from collections.abc import Iterable
  from functools import partial
  from math import ceil
  from operator import getitem
  from threading import Lock
  from typing import TYPE_CHECKING, Literal, overload
  import numpy
  import pandas
  import dask.array
  from dask.base import is_dask_collection, tokenize
  from dask.blockwise import BlockwiseDepDict, blockwise
  from dask.dataframe._compat import is_any_real_numeric_dtype
  from dask.dataframe.backends import dataframe_creation_dispatch
  from dask.dataframe.core import DataFrame, Index, Series, _concat, _emulate, _Frame, apply_and_enforce, has_parallel_type, new_dd_object
  from dask.dataframe.dispatch import meta_lib_from_array
  from dask.dataframe.io.utils import DataFrameIOFunction
  from dask.dataframe.utils import check_meta, insert_meta_param_description, is_series_like, make_meta
  from dask.delayed import Delayed, delayed
  from dask.highlevelgraph import HighLevelGraph
  from dask.layers import DataFrameIOLayer
  from dask.utils import M, funcname, is_arraylike
label_desc: Use image node_id 8 for calling a global function with example usage: from_dask_array(x, columns, index, meta) and returns: new_dd_object