ValueError: Unrecognized filesystem type in URI: abfss

[djouallah]

Apache Iceberg version

None

Please describe the bug 🐞

reproducible example
https://colab.research.google.com/drive/1EjffJO75-8Rj4V0MGKUsoFHDOGgicKgK#scrollTo=8WRyLlmyXnXu

[kevinjqliu]

Pulling the example out of the notebook:

table = catalog.create_table(
    "default.taxi_dataset",
    schema=df.schema,location="abfss://onelakene.dfs.core.windows.net/aemo/iceberg"
)

error:

WARNING:pyiceberg.io:Could not initialize FileIO: pyiceberg.io.fsspec.FsspecFileIO
---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
<ipython-input-4-f052e36c7671> in <cell line: 1>()
----> 1 table = catalog.create_table(
      2     "default.taxi_dataset",
      3     schema=df.schema,location="abfss://onelakene.dfs.core.windows.net/aemo/iceberg"
      4 )

3 frames
/usr/local/lib/python3.10/dist-packages/pyiceberg/io/pyarrow.py in _initialize_fs(self, scheme, netloc)
    390             return PyArrowLocalFileSystem()
    391         else:
--> 392             raise ValueError(f"Unrecognized filesystem type in URI: {scheme}")
    393 
    394     def new_input(self, location: str) -> PyArrowFile:

ValueError: Unrecognized filesystem type in URI: abfss

[kevinjqliu]

abfss isn't currently supported in pyarrow FS implementation

iceberg-python/pyiceberg/io/pyarrow.py

Lines 339 to 393 in 7f712fd

	if scheme in {"s3", "s3a", "s3n"}:
	from pyarrow.fs import S3FileSystem
	client_kwargs: Dict[str, Any] = {
	"endpoint_override": self.properties.get(S3_ENDPOINT),
	"access_key": self.properties.get(S3_ACCESS_KEY_ID),
	"secret_key": self.properties.get(S3_SECRET_ACCESS_KEY),
	"session_token": self.properties.get(S3_SESSION_TOKEN),
	"region": self.properties.get(S3_REGION),
	}
	if proxy_uri := self.properties.get(S3_PROXY_URI):
	client_kwargs["proxy_options"] = proxy_uri
	if connect_timeout := self.properties.get(S3_CONNECT_TIMEOUT):
	client_kwargs["connect_timeout"] = float(connect_timeout)
	return S3FileSystem(**client_kwargs)
	elif scheme == "hdfs":
	from pyarrow.fs import HadoopFileSystem
	hdfs_kwargs: Dict[str, Any] = {}
	if netloc:
	return HadoopFileSystem.from_uri(f"hdfs://{netloc}")
	if host := self.properties.get(HDFS_HOST):
	hdfs_kwargs["host"] = host
	if port := self.properties.get(HDFS_PORT):
	# port should be an integer type
	hdfs_kwargs["port"] = int(port)
	if user := self.properties.get(HDFS_USER):
	hdfs_kwargs["user"] = user
	if kerb_ticket := self.properties.get(HDFS_KERB_TICKET):
	hdfs_kwargs["kerb_ticket"] = kerb_ticket
	return HadoopFileSystem(**hdfs_kwargs)
	elif scheme in {"gs", "gcs"}:
	from pyarrow.fs import GcsFileSystem
	gcs_kwargs: Dict[str, Any] = {}
	if access_token := self.properties.get(GCS_TOKEN):
	gcs_kwargs["access_token"] = access_token
	if expiration := self.properties.get(GCS_TOKEN_EXPIRES_AT_MS):
	gcs_kwargs["credential_token_expiration"] = millis_to_datetime(int(expiration))
	if bucket_location := self.properties.get(GCS_DEFAULT_LOCATION):
	gcs_kwargs["default_bucket_location"] = bucket_location
	if endpoint := self.properties.get(GCS_ENDPOINT):
	url_parts = urlparse(endpoint)
	gcs_kwargs["scheme"] = url_parts.scheme
	gcs_kwargs["endpoint_override"] = url_parts.netloc
	return GcsFileSystem(**gcs_kwargs)
	elif scheme == "file":
	return PyArrowLocalFileSystem()
	else:
	raise ValueError(f"Unrecognized filesystem type in URI: {scheme}")

but it is available in the fsspec implementation

iceberg-python/pyiceberg/io/fsspec.py

Lines 181 to 182 in 7f712fd

	"abfs": _adlfs,
	"abfss": _adlfs,

[kevinjqliu]

Looks like pyarrow can support "fsspec-compatible filesystems" like Azure Blob Storage (abfs/abfss)
https://arrow.apache.org/docs/python/filesystems.html#using-fsspec-compatible-filesystems-with-arrow

There's an issue open to make fsspec and pyarrow filesystems cross-compatible #310

[kevinjqliu]

In the meantime, I think you might be able to workaround this by explicitly using fsspec.
You'd have to set this in the catalog properties setting

catalog = SqlCatalog(
    "default",
    **{
        "uri": "sqlite:///:memory:",
        "adlfs.account-name": userdata.get("account_name") ,
        "adlfs.account-key": userdata.get ("AZURE_STORAGE_ACCOUNT_KEY"),
        "adlfs.tenant-id" : userdata.get("azure_storage_tenant_id"),
        "py-io-impl": "pyiceberg.io.fsspec.FsspecFileIO"
    },
)

iceberg-python/pyiceberg/catalog/sql.py

Line 185 in 7f712fd

io = load_file_io(properties=self.properties, location=metadata_location)

[djouallah]

getting this error now

319             return file_io
 320         else:
--> 321             raise ValueError(f"Could not initialize FileIO: {io_impl}")
 322 
 323     # Check the table location

ValueError: Could not initialize FileIO: pyiceberg.io.fsspec.FsspecFileIO

[kevinjqliu]

oh interesting, this is because of a dependency issue.

The actual error shows up when you try to import that class

from pyiceberg.io.fsspec import FsspecFileIO

fsspec has a dependency on botocore and botocore is not installed with !pip install -q pyiceberg[adlfs]

iceberg-python/pyiceberg/io/fsspec.py

Lines 33 to 34 in 7f712fd

	from botocore import UNSIGNED
	from botocore.awsrequest import AWSRequest

To resolve this, install botocore

!pip install botocore

In the future, we'll get rid of this dependency issue. This is caught by deptry in #528.

[djouallah]

Thanks, it works now